优化查询性能

database/records/projection.go

@@ -0,0 +1,63 @@
+package records
+
+import "github.com/komari-monitor/komari/database/models"
+
+// LoadProjection is the field subset selected by a load_type request.
+type LoadProjection struct {
+	Cpu            *float32 `json:"cpu,omitempty"`
+	Gpu            *float32 `json:"gpu,omitempty"`
+	Ram            *int64   `json:"ram,omitempty"`
+	RamTotal       *int64   `json:"ram_total,omitempty"`
+	Swap           *int64   `json:"swap,omitempty"`
+	SwapTotal      *int64   `json:"swap_total,omitempty"`
+	Load           *float32 `json:"load,omitempty"`
+	Temp           *float32 `json:"temp,omitempty"`
+	Disk           *int64   `json:"disk,omitempty"`
+	DiskTotal      *int64   `json:"disk_total,omitempty"`
+	NetIn          *int64   `json:"net_in,omitempty"`
+	NetOut         *int64   `json:"net_out,omitempty"`
+	NetTotalUp     *int64   `json:"net_total_up,omitempty"`
+	NetTotalDown   *int64   `json:"net_total_down,omitempty"`
+	Process        *int     `json:"process,omitempty"`
+	Connections    *int     `json:"connections,omitempty"`
+	ConnectionsUdp *int     `json:"connections_udp,omitempty"`
+}
+
+// ProjectLoad returns the load_type projection for one record.
+func ProjectLoad(r models.Record, loadType string) LoadProjection {
+	var p LoadProjection
+	switch loadType {
+	case "cpu":
+		p.Cpu = ptr(r.Cpu)
+	case "gpu":
+		p.Gpu = ptr(r.Gpu)
+	case "ram":
+		p.Ram = ptr(r.Ram)
+		p.RamTotal = ptr(r.RamTotal)
+	case "swap":
+		p.Swap = ptr(r.Swap)
+		p.SwapTotal = ptr(r.SwapTotal)
+	case "load":
+		p.Load = ptr(r.Load)
+	case "temp":
+		p.Temp = ptr(r.Temp)
+	case "disk":
+		p.Disk = ptr(r.Disk)
+		p.DiskTotal = ptr(r.DiskTotal)
+	case "network":
+		p.NetIn = ptr(r.NetIn)
+		p.NetOut = ptr(r.NetOut)
+		p.NetTotalUp = ptr(r.NetTotalUp)
+		p.NetTotalDown = ptr(r.NetTotalDown)
+	case "process":
+		p.Process = ptr(r.Process)
+	case "connections":
+		p.Connections = ptr(r.Connections)
+		p.ConnectionsUdp = ptr(r.ConnectionsUdp)
+	default:
+		p.Cpu = ptr(r.Cpu)
+	}
+	return p
+}
+
+func ptr[T any](v T) *T { return &v }

database/records/records.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"time"
 
-	"github.com/komari-monitor/komari/internal/metricstore"
 	"github.com/komari-monitor/komari/database/models"
+	"github.com/komari-monitor/komari/internal/metricstore"
 )
 
 // 历史监控数据已完全迁移到 metric store（默认 SQLite ./data/metrics.db，或配置的
@@ -19,21 +19,77 @@ func DeleteAll() error {
 	return metricstore.DeleteAllRecords(context.Background())
 }
 
-// GetGPURecordsByClientAndTime 获取 GPU 记录数据。
+// QueryGPURecords returns GPU records using the caller's cancellation context.
+func QueryGPURecords(ctx context.Context, uuid string, start, end time.Time) ([]models.GPURecord, error) {
+	return metricstore.GetGPURecordsByClientAndTime(ctx, uuid, start, end)
+}
+
+// GetGPURecordsByClientAndTime is a compatibility-only wrapper without cancellation.
+// New request handlers should use QueryGPURecords.
 func GetGPURecordsByClientAndTime(uuid string, start, end time.Time) ([]models.GPURecord, error) {
-	return metricstore.GetGPURecordsByClientAndTime(context.Background(), uuid, start, end)
+	return QueryGPURecords(context.Background(), uuid, start, end)
 }
 
-func GetRecordsByClientAndTime(uuid string, start, end time.Time) ([]models.Record, error) {
-	return metricstore.GetRecordsByClientAndTime(context.Background(), uuid, start, end)
+// RecordQueryOptions narrows metric selection and the response point budget.
+type RecordQueryOptions = metricstore.RecordQueryOptions
+
+// LoadTypeMetricNames maps the RPC load_type to the stored metrics that back
+// that field. An empty result means "all load metrics" (including temp and
+// unknown values, which stay on the compatible full-query path).
+func LoadTypeMetricNames(loadType string) []string {
+	var names []string
+	switch loadType {
+	case "cpu":
+		return []string{metricstore.MetricCPU}
+	case "gpu":
+		names = []string{metricstore.MetricGPU}
+	case "ram":
+		names = []string{metricstore.MetricRAM}
+	case "swap":
+		names = []string{metricstore.MetricSwap}
+	case "load":
+		names = []string{metricstore.MetricLoad}
+	case "disk":
+		names = []string{metricstore.MetricDisk}
+	case "network":
+		names = []string{metricstore.MetricNetIn, metricstore.MetricNetOut, metricstore.MetricNetTotalUp, metricstore.MetricNetTotalDown}
+	case "process":
+		names = []string{metricstore.MetricProcess}
+	case "connections":
+		names = []string{metricstore.MetricConnections, metricstore.MetricConnectionsUDP}
+	default:
+		return nil
+	}
+	// CPU is the common presence anchor: the previous implementation built one
+	// record per timestamp where any load metric existed. Including CPU keeps
+	// those timestamps (and zero values for a missing selected metric) without
+	// going back to all fifteen metrics.
+	return append(names, metricstore.MetricCPU)
 }
 
-// GetRecordMetricMaxByClientAndTime 获取单项监控指标在各时间桶内的最大值。
+// QueryRecords returns one client's records, or all clients when uuid is empty.
+// Request handlers should use this context-aware entry point.
+func QueryRecords(ctx context.Context, uuid string, start, end time.Time, opts ...RecordQueryOptions) ([]models.Record, error) {
+	if uuid == "" {
+		return metricstore.GetRecordsByTime(ctx, start, end, opts...)
+	}
+	return metricstore.GetRecordsByClientAndTime(ctx, uuid, start, end, opts...)
+}
+
+// GetRecordsByClientAndTime is a compatibility-only wrapper without cancellation.
+// New request handlers should use QueryRecords.
+func GetRecordsByClientAndTime(uuid string, start, end time.Time, opts ...RecordQueryOptions) ([]models.Record, error) {
+	return metricstore.GetRecordsByClientAndTime(context.Background(), uuid, start, end, opts...)
+}
+
+// GetRecordMetricMaxByClientAndTime is a compatibility-only wrapper without cancellation.
+// New request handlers should use metricstore.GetRecordMetricMaxByClientAndTime.
 func GetRecordMetricMaxByClientAndTime(uuid, metricName string, start, end time.Time) ([]models.Record, error) {
 	return metricstore.GetRecordMetricMaxByClientAndTime(context.Background(), uuid, metricName, start, end)
 }
 
-// GetRecordsByTime 获取所有客户端在时间范围内的记录。
-func GetRecordsByTime(start, end time.Time) ([]models.Record, error) {
-	return metricstore.GetRecordsByTime(context.Background(), start, end)
+// GetRecordsByTime is a compatibility-only wrapper without cancellation.
+// New request handlers should use QueryRecords with an empty uuid.
+func GetRecordsByTime(start, end time.Time, opts ...RecordQueryOptions) ([]models.Record, error) {
+	return metricstore.GetRecordsByTime(context.Background(), start, end, opts...)
 }

database/tasks/ping.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"slices"
 	"sort"
-	"time"
 
 	"github.com/komari-monitor/komari/database/dbcore"
 	"github.com/komari-monitor/komari/database/models"
@@ -190,7 +189,3 @@ func AddDefaultOnClientUUID(uuid string) error {
 	}
 	return nil
 }
-
-func GetPingRecords(uuid string, taskId int, start, end time.Time) ([]models.PingRecord, error) {
-	return metricstore.GetPingRecords(context.Background(), uuid, taskId, start, end)
-}

database/tasks/ping_summary.go

@@ -0,0 +1,180 @@
+package tasks
+
+import (
+	"math"
+	"sort"
+	"time"
+
+	"github.com/komari-monitor/komari/database/models"
+	"github.com/komari-monitor/komari/internal/sampling"
+)
+
+// PingClientSummary is the per-client aggregate shared by ping RPC adapters.
+type PingClientSummary struct {
+	Total       int
+	Loss        int
+	Min         int
+	Max         int
+	LossPercent float64
+}
+
+// PingTaskSummary is the per-task aggregate shared by ping RPC adapters.
+type PingTaskSummary struct {
+	Total       int
+	Loss        int
+	Valid       int
+	Min         int
+	Max         int
+	Avg         int
+	Latest      int
+	P50         int
+	P99         int
+	LossPercent float64
+	P99P50Ratio float64
+}
+
+// PingSummary aggregates records once so legacy and public ping handlers only
+// differ in response shape.
+type PingSummary struct {
+	Clients map[string]PingClientSummary
+	Tasks   map[uint]PingTaskSummary
+}
+
+type pingClientAccumulator struct {
+	total, loss, min, max int
+}
+
+type pingTaskAccumulator struct {
+	total, loss, min, max, sum, valid int
+	latest                            int
+	latestTime                        time.Time
+	latencies                         []int
+}
+
+// SummarizePingRecords computes client and task aggregates in one pass.
+func SummarizePingRecords(recs []models.PingRecord) PingSummary {
+	clients := make(map[string]*pingClientAccumulator)
+	taskMap := make(map[uint]*pingTaskAccumulator)
+	for _, r := range recs {
+		cs := clients[r.Client]
+		if cs == nil {
+			cs = &pingClientAccumulator{}
+			clients[r.Client] = cs
+		}
+		cs.total++
+		ts := taskMap[r.TaskId]
+		if ts == nil {
+			ts = &pingTaskAccumulator{latest: -1}
+			taskMap[r.TaskId] = ts
+		}
+		ts.total++
+		if r.Value < 0 {
+			cs.loss++
+			ts.loss++
+			continue
+		}
+		if cs.min == 0 || r.Value < cs.min {
+			cs.min = r.Value
+		}
+		if r.Value > cs.max {
+			cs.max = r.Value
+		}
+		ts.valid++
+		ts.sum += r.Value
+		if ts.min == 0 || r.Value < ts.min {
+			ts.min = r.Value
+		}
+		if r.Value > ts.max {
+			ts.max = r.Value
+		}
+		if ts.latestTime.IsZero() || r.Time.After(ts.latestTime) {
+			ts.latestTime = r.Time
+			ts.latest = r.Value
+		}
+		ts.latencies = append(ts.latencies, r.Value)
+	}
+
+	out := PingSummary{Clients: make(map[string]PingClientSummary, len(clients)), Tasks: make(map[uint]PingTaskSummary, len(taskMap))}
+	for client, v := range clients {
+		loss := 0.0
+		if v.total > 0 {
+			loss = float64(v.loss) / float64(v.total) * 100
+		}
+		out.Clients[client] = PingClientSummary{Total: v.total, Loss: v.loss, Min: v.min, Max: v.max, LossPercent: loss}
+	}
+	for id, v := range taskMap {
+		loss := 0.0
+		if v.total > 0 {
+			loss = float64(v.loss) / float64(v.total) * 100
+		}
+		avg := 0
+		if v.valid > 0 {
+			avg = v.sum / v.valid
+		}
+		p50, p99 := pingPercentiles(v.latencies)
+		ratio := 0.0
+		if p50 > 0 && p99 >= p50 {
+			adjustedBase := math.Max(math.Min(float64(p50), 50.0), 10.0)
+			ratio = float64(p99-p50) / adjustedBase
+		}
+		out.Tasks[id] = PingTaskSummary{
+			Total: v.total, Loss: v.loss, Valid: v.valid, Min: v.min, Max: v.max,
+			Avg: avg, Latest: v.latest, P50: p50, P99: p99, LossPercent: loss, P99P50Ratio: ratio,
+		}
+	}
+	return out
+}
+
+// SamplePingRecordsByTask applies the legacy ping maxCount policy: allocate
+// points proportionally by task, sample each task evenly and return ascending
+// by time. Statistics must be computed from the unsampled records.
+func SamplePingRecordsByTask(recs []models.PingRecord, maxCount int) []models.PingRecord {
+	if maxCount <= 0 || len(recs) <= maxCount {
+		return recs
+	}
+	groups := make(map[uint][]models.PingRecord)
+	for _, r := range recs {
+		groups[r.TaskId] = append(groups[r.TaskId], r)
+	}
+	meta := make([]sampling.Group[uint], 0, len(groups))
+	for id, points := range groups {
+		sort.Slice(points, func(i, j int) bool { return points[i].Time.Before(points[j].Time) })
+		groups[id] = points
+		meta = append(meta, sampling.Group[uint]{Key: id, Length: len(points)})
+	}
+	targets := sampling.AllocateTargets(meta, maxCount)
+	out := make([]models.PingRecord, 0, maxCount)
+	for id, points := range groups {
+		out = append(out, sampling.SampleEvenly(points, targets[id])...)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Time.Before(out[j].Time) })
+	return out
+}
+
+func pingPercentiles(values []int) (int, int) {
+	if len(values) == 0 {
+		return 0, 0
+	}
+	sort.Ints(values)
+	return pingPercentile(values, 0.50), pingPercentile(values, 0.99)
+}
+
+func pingPercentile(values []int, percentile float64) int {
+	if len(values) == 0 {
+		return 0
+	}
+	if percentile <= 0 {
+		return values[0]
+	}
+	if percentile >= 1 {
+		return values[len(values)-1]
+	}
+	pos := float64(len(values)-1) * percentile
+	lo := int(math.Floor(pos))
+	hi := int(math.Ceil(pos))
+	if lo == hi {
+		return values[lo]
+	}
+	frac := pos - float64(lo)
+	return int(math.Round(float64(values[lo]) + (float64(values[hi])-float64(values[lo]))*frac))
+}

internal/metricstore/legacy_records.go

@@ -10,14 +10,32 @@ import (
 	"github.com/komari-monitor/komari/pkg/metric"
 )
 
-// GetRecordsByClientAndTime 从 metric store 查询记录并重构为 models.Record
-func GetRecordsByClientAndTime(ctx context.Context, clientUUID string, start, end time.Time) ([]models.Record, error) {
+// RecordQueryOptions narrows the record reconstruction work. MetricNames
+// limits which stored load metrics are queried; empty means all load metrics.
+// MaxPoints is a total response-point budget; it is applied before aggregation,
+// so an all-client query does not materialize thousands of points per client
+// only to sample them afterwards.
+type RecordQueryOptions struct {
+	MetricNames []string
+	MaxPoints   int
+}
+
+// GetRecordsByClientAndTime 从 metric store 查询记录并重构为 models.Record。
+// opts 为空表示兼容旧行为；有值时用于指标选择和点数预算。
+func GetRecordsByClientAndTime(ctx context.Context, clientUUID string, start, end time.Time, opts ...RecordQueryOptions) ([]models.Record, error) {
 	s := GetStore()
 	if s == nil {
 		return nil, fmt.Errorf("metric store not enabled")
 	}
 
-	return getRecordsByClientAndTimeFromSeries(ctx, s, clientUUID, start, end)
+	return getRecordsFromSeries(ctx, s, clientUUID, start, end, false, firstRecordOptions(opts))
+}
+
+func firstRecordOptions(opts []RecordQueryOptions) RecordQueryOptions {
+	if len(opts) == 0 {
+		return RecordQueryOptions{}
+	}
+	return opts[0]
 }
 
 // GetRecordMetricMaxByClientAndTime 查询单项监控指标在各时间桶内的最大值。
@@ -30,28 +48,17 @@ func GetRecordMetricMaxByClientAndTime(ctx context.Context, clientUUID, recordMe
 	return getRecordMetricMaxByClientAndTimeFromSeries(ctx, s, clientUUID, recordMetric, start, end)
 }
 
-// GetRecordsByTime 从 metric store 查询所有客户端在时间范围内的记录
-func GetRecordsByTime(ctx context.Context, start, end time.Time) ([]models.Record, error) {
+// GetRecordsByTime 从 metric store 查询所有客户端在时间范围内的记录。
+// opts 为空表示兼容旧行为；有值时用于指标选择和总点数预算。
+func GetRecordsByTime(ctx context.Context, start, end time.Time, opts ...RecordQueryOptions) ([]models.Record, error) {
 	s := GetStore()
 	if s == nil {
 		return nil, fmt.Errorf("metric store not enabled")
 	}
 
-	interval := recordSeriesInterval(s, start, end, time.Now().UTC())
-	entityIDs, err := listRecordEntityIDs(ctx, s, start, end, interval)
-	if err != nil {
-		return nil, err
-	}
-	var records []models.Record
-	for _, entityID := range entityIDs {
-		items, err := getRecordsByClientAndTimeFromSeries(ctx, s, entityID, start, end)
-		if err != nil {
-			return nil, err
-		}
-		records = append(records, items...)
-	}
-	sortRecords(records)
-	return records, nil
+	// Query all entities together. Entity-only grouping retains the old per-client
+	// aggregation across tags without repeating the tier scan for every client.
+	return getRecordsFromSeries(ctx, s, "", start, end, true, firstRecordOptions(opts))
 }
 
 type recordSeriesKey struct {
@@ -59,30 +66,54 @@ type recordSeriesKey struct {
 	ts     int64
 }
 
-func getRecordsByClientAndTimeFromSeries(ctx context.Context, s *metric.Store, clientUUID string, start, end time.Time) ([]models.Record, error) {
-	now := time.Now().UTC()
-	interval := recordSeriesInterval(s, start, end, now)
-	recordMap := make(map[recordSeriesKey]*models.Record)
-
-	specs := make([]metric.BatchSeriesSpec, 0, len(loadRecordMetricNames))
-	for _, metricName := range loadRecordMetricNames {
-		specs = append(specs, metric.BatchSeriesSpec{
-			MetricName:   metricName,
-			Aggregations: []metric.Aggregation{recordMetricAggregation(metricName)},
-			Interval:     interval,
-		})
+func getRecordsFromSeries(ctx context.Context, s *metric.Store, clientUUID string, start, end time.Time, allEntities bool, opts RecordQueryOptions) ([]models.Record, error) {
+	metricNames := opts.MetricNames
+	if len(metricNames) == 0 {
+		metricNames = loadRecordMetricNames
 	}
-	loaded, err := s.SeriesBatch(ctx, metric.BatchSeriesQuery{
-		Specs:     specs,
-		EntityIDs: []string{clientUUID},
-		Start:     start,
-		End:       end,
-		Order:     metric.OrderAsc,
-	}, now)
+	specs := make([]MetricSpec, 0, len(metricNames))
+	for _, name := range metricNames {
+		specs = append(specs, MetricSpec{Name: name, Aggregations: []metric.Aggregation{recordMetricAggregation(name)}})
+	}
+	entities := EntityScope{All: allEntities}
+	grouping := GroupMerged
+	if allEntities {
+		grouping = GroupEntity
+	} else {
+		entities.IDs = []string{clientUUID}
+	}
+	rounding := RoundFloor
+	if opts.MaxPoints > 0 {
+		rounding = RoundCeil
+	}
+	loaded, err := QueryMetrics(ctx, MetricsQuery{
+		Store: s, Mode: QuerySeries, Source: SourceRollup,
+		OmitTags: true, OmitLabels: true, CompactDefinitions: true,
+		Entities: entities, Specs: specs, Start: start, End: end,
+		Order: metric.OrderAsc, Grouping: grouping,
+		Budget:   PointBudget{PerSeries: 500, Total: opts.MaxPoints},
+		Rounding: rounding, AllowMissing: true,
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to query metrics: %w", err)
 	}
-	for metricName, aggMap := range loaded.Values {
+	return assembleRecordsFromSeries(loaded.Values, clientUUID), nil
+}
+
+// assembleRecordsFromSeries unions metric buckets without allocating a record per key.
+func assembleRecordsFromSeries(values map[string]map[metric.Aggregation][]metric.AggregatePoint, clientUUID string) []models.Record {
+	// Aligned metrics normally share buckets: the largest selected metric is a
+	// conservative capacity estimate, not the sum of the metric fanout. Missing
+	// buckets can grow the union beyond it (and duplicate seconds can shrink it).
+	capacity := 0
+	for name, aggregations := range values {
+		if n := len(aggregations[recordMetricAggregation(name)]); n > capacity {
+			capacity = n
+		}
+	}
+	records := make([]models.Record, 0, capacity)
+	recordIndex := make(map[recordSeriesKey]int, capacity)
+	for metricName, aggMap := range values {
 		agg := recordMetricAggregation(metricName)
 		points, ok := aggMap[agg]
 		if !ok {
@@ -94,22 +125,22 @@ func getRecordsByClientAndTimeFromSeries(ctx context.Context, s *metric.Store, c
 				entityID = clientUUID
 			}
 			key := recordSeriesKey{client: entityID, ts: point.Bucket.Unix()}
-			if recordMap[key] == nil {
-				recordMap[key] = &models.Record{
+			index, exists := recordIndex[key]
+			if !exists {
+				index = len(records)
+				recordIndex[key] = index
+				records = append(records, models.Record{
 					Client: entityID,
 					Time:   point.Bucket.UTC(),
-				}
+				})
 			}
-			applyRecordMetricValue(recordMap[key], metricName, point.Value)
+			// Resolve the address after append: the backing array may have moved.
+			applyRecordMetricValue(&records[index], metricName, point.Value)
 		}
 	}
 
-	records := make([]models.Record, 0, len(recordMap))
-	for _, rec := range recordMap {
-		records = append(records, *rec)
-	}
 	sortRecords(records)
-	return records, nil
+	return records
 }
 
 func getRecordMetricMaxByClientAndTimeFromSeries(ctx context.Context, s *metric.Store, clientUUID, recordMetric string, start, end time.Time) ([]models.Record, error) {
@@ -118,22 +149,19 @@ func getRecordMetricMaxByClientAndTimeFromSeries(ctx context.Context, s *metric.
 		return nil, fmt.Errorf("unsupported record metric %q", recordMetric)
 	}
 
-	now := time.Now().UTC()
-	points, err := s.Series(ctx, metric.AggregateQuery{
-		Query: metric.Query{
-			MetricName: metricName,
-			EntityID:   clientUUID,
-			Start:      start,
-			End:        end,
-			Order:      metric.OrderAsc,
-		},
-		Aggregation: metric.AggMax,
-		Interval:    recordSeriesInterval(s, start, end, now),
-	}, now)
+	loaded, err := QueryMetrics(ctx, MetricsQuery{
+		Store: s, Mode: QuerySeries, Source: SourceRollup,
+		OmitTags: true, OmitLabels: true, CompactDefinitions: true,
+		Entities: EntityScope{IDs: []string{clientUUID}},
+		Specs:    []MetricSpec{{Name: metricName, Aggregations: []metric.Aggregation{metric.AggMax}}},
+		Start:    start, End: end, Order: metric.OrderAsc, Grouping: GroupMerged,
+		Budget: PointBudget{PerSeries: 500}, Rounding: RoundFloor, AllowMissing: true,
+	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to query metric %s: %w", metricName, err)
 	}
 
+	points := loaded.Values[metricName][metric.AggMax]
 	records := make([]models.Record, 0, len(points))
 	for _, point := range points {
 		entityID := point.EntityID
@@ -158,45 +186,6 @@ func recordMetricAggregation(metricName string) metric.Aggregation {
 	}
 }
 
-func recordSeriesInterval(s *metric.Store, start, end, now time.Time) time.Duration {
-	interval := downsampleInterval(end.Sub(start), 500)
-	return s.CompatibleSeriesInterval(start, now, interval)
-}
-
-// downsampleInterval returns the standard interval that keeps rangeDuration
-// within about maxPoints buckets, never finer than one second.
-func downsampleInterval(rangeDuration time.Duration, maxPoints int) time.Duration {
-	points := time.Duration(maxPoints)
-	interval := (rangeDuration + points - 1) / points
-	if interval < time.Second {
-		return time.Second
-	}
-	return metric.FloorStandardInterval(interval)
-}
-
-func listRecordEntityIDs(ctx context.Context, s *metric.Store, start, end time.Time, interval time.Duration) ([]string, error) {
-	seen := make(map[string]struct{})
-	for _, metricName := range loadRecordMetricNames {
-		ids, err := s.EntityIDs(ctx, metric.Query{
-			MetricName: metricName,
-			Start:      start.Add(-interval),
-			End:        end,
-		})
-		if err != nil {
-			return nil, err
-		}
-		for _, id := range ids {
-			seen[id] = struct{}{}
-		}
-	}
-	out := make([]string, 0, len(seen))
-	for id := range seen {
-		out = append(out, id)
-	}
-	sort.Strings(out)
-	return out, nil
-}
-
 func applyRecordMetricValue(rec *models.Record, metricName string, value float64) {
 	switch metricName {
 	case MetricCPU:
@@ -258,17 +247,21 @@ func GetGPURecordsByClientAndTime(ctx context.Context, clientUUID string, start,
 	}
 	recordMap := make(map[gpuKey]*models.GPURecord)
 
+	specs := make([]MetricSpec, 0, len(gpuMetrics))
+	for _, name := range gpuMetrics {
+		specs = append(specs, MetricSpec{Name: name})
+	}
+	loaded, err := QueryMetrics(ctx, MetricsQuery{
+		Store: s, Mode: QueryPoints, Source: SourceRaw,
+		OmitLabels: true, CompactDefinitions: true,
+		Entities: EntityScope{IDs: []string{clientUUID}}, Specs: specs,
+		Start: start, End: end, Order: metric.OrderAsc, AllowMissing: true,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to query GPU metrics: %w", err)
+	}
 	for _, metricName := range gpuMetrics {
-		points, err := s.Query(ctx, metric.Query{
-			MetricName: metricName,
-			EntityID:   clientUUID,
-			Start:      start,
-			End:        end,
-			Order:      metric.OrderAsc,
-		})
-		if err != nil {
-			continue // GPU 数据可能不存在
-		}
+		points := loaded.Raw[metricName]
 
 		for _, p := range points {
 			deviceIndex := 0

internal/metricstore/ping_records.go

@@ -69,33 +69,26 @@ func GetPingRecords(ctx context.Context, clientUUID string, taskID int, start, e
 		return nil, fmt.Errorf("metric store not enabled")
 	}
 
-	query := metric.Query{
-		MetricName: MetricPingLatency,
-		Start:      start,
-		End:        end,
-		Order:      metric.OrderAsc,
+	query := MetricsQuery{
+		Store: s, Mode: QuerySeries, Source: SourceRollup,
+		CompactDefinitions: true,
+		Entities:           EntityScope{All: clientUUID == ""},
+		Specs:              []MetricSpec{{Name: MetricPingLatency, Aggregations: []metric.Aggregation{metric.AggLast}}},
+		Start:              start, End: end, Order: metric.OrderAsc, Grouping: GroupSeries,
+		Budget: PointBudget{PerSeries: 4000}, Rounding: RoundFloor, AllowMissing: true,
 	}
-
 	if clientUUID != "" {
-		query.EntityID = clientUUID
+		query.Entities.IDs = []string{clientUUID}
 	}
-
 	if taskID >= 0 {
 		query.Tags = map[string]string{"task_id": fmt.Sprintf("%d", taskID)}
 	}
-
-	interval := downsampleInterval(end.Sub(start), 4000)
-	interval = s.CompatibleSeriesInterval(start, time.Now().UTC(), interval)
-	points, err := s.Series(ctx, metric.AggregateQuery{
-		Query:          query,
-		Aggregation:    metric.AggLast,
-		Interval:       interval,
-		PreserveSeries: true,
-	}, time.Now().UTC())
+	loaded, err := QueryMetrics(ctx, query)
 	if err != nil {
 		return nil, err
 	}
 
+	points := loaded.Values[MetricPingLatency][metric.AggLast]
 	records := make([]models.PingRecord, 0, len(points))
 	for _, p := range points {
 		taskIDVal := uint(0)

internal/metricstore/query.go

@@ -0,0 +1,766 @@
+package metricstore
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/komari-monitor/komari/pkg/metric"
+)
+
+// QueryMode selects exact samples, adaptive series, or exclusive-boundary lookups.
+type QueryMode string
+
+const (
+	QueryPoints       QueryMode = "points"
+	QuerySeries       QueryMode = "series"
+	QueryLatestBefore QueryMode = "latest_before"
+)
+
+type QuerySource string
+
+const (
+	SourceAuto   QuerySource = "auto"
+	SourceRaw    QuerySource = "raw"
+	SourceRollup QuerySource = "rollup"
+)
+
+type QueryGrouping string
+
+const (
+	GroupMerged QueryGrouping = "merged"
+	GroupEntity QueryGrouping = "entity"
+	GroupSeries QueryGrouping = "series"
+)
+
+type IntervalRounding string
+
+const (
+	RoundFloor IntervalRounding = "floor"
+	RoundCeil  IntervalRounding = "ceil"
+)
+
+// EntityScope deliberately distinguishes an empty authorized set from all entities.
+// All and IDs are mutually exclusive. IDs are deduplicated without modifying callers.
+type EntityScope struct {
+	All bool
+	IDs []string
+}
+type MetricSpec struct {
+	Name         string
+	Aggregations []metric.Aggregation
+	// Positive intervals are explicit and are preserved, even for historical data.
+	Interval time.Duration
+	// Zero inherits Budget.PerSeries, then 500. -1 disables adaptive downsampling:
+	// a zero Interval then selects the finest compatible tier, not a whole-window bucket.
+	MaxPoints int
+}
+
+// PointBudget controls adaptive interval planning, not a raw-row SQL LIMIT.
+// Total divides the point target across entities (metrics may be columns of one
+// legacy record); it is not a hard cap across metrics, aggregations or tag variants.
+// Raw queries and explicit intervals retain their exact historical semantics.
+// Response adapters can impose their existing final sampling policy separately.
+type PointBudget struct {
+	Total     int
+	PerSeries int
+}
+type FillPolicy struct {
+	Gaps               bool
+	PingNegativeAsNull bool
+}
+
+// QueryProjection controls retained result representations. The default keeps all
+// source data for compatibility; chart requires Fill and retains only Filled.
+type QueryProjection string
+
+const (
+	ProjectionFull  QueryProjection = "full"
+	ProjectionChart QueryProjection = "chart"
+)
+
+type MetricsQuery struct {
+	// Output omissions are opt-in; filtering and aggregation identity are unchanged.
+	OmitTags           bool
+	OmitLabels         bool
+	CompactDefinitions bool
+	Projection         QueryProjection
+	Store              *metric.Store
+	Mode               QueryMode
+	Source             QuerySource
+	Entities           EntityScope
+	Specs              []MetricSpec
+	Start, End         time.Time
+	Tags               map[string]string
+	Order              metric.Order
+	Grouping           QueryGrouping
+	Budget             PointBudget
+	Rounding           IntervalRounding
+	Fill               FillPolicy
+	AllowMissing       bool
+	Now                time.Time
+}
+type MetricsResult struct {
+	Definitions map[string]metric.Definition
+	Raw         map[string][]metric.Point
+	Values      map[string]map[metric.Aggregation][]metric.AggregatePoint
+	Intervals   map[string]time.Duration
+	IsRaw       bool
+	Filled      []MetricSeries
+	// Latest is keyed by metric name, then entity; missing samples are absent.
+	Latest map[string]map[string]metric.Point
+}
+type MetricSeries struct {
+	MetricName, EntityID string
+	Aggregation          metric.Aggregation
+	Tags                 map[string]string
+	Interval             time.Duration
+	Points               []MetricValue
+}
+type MetricValue struct {
+	Time         time.Time
+	Value        *float64
+	Count        int
+	Tags, Labels map[string]string
+}
+
+// DownsampleInterval returns a standard interval that keeps rangeDuration
+// within about maxPoints buckets, never finer than one second. roundUp is used
+// by point budgets: flooring can exceed the budget, while compatibility queries
+// preserve the historical floor behavior.
+func DownsampleInterval(rangeDuration time.Duration, maxPoints int, roundUp bool) time.Duration {
+	if maxPoints < 1 {
+		maxPoints = 1
+	}
+	points := time.Duration(maxPoints)
+	interval := (rangeDuration + points - 1) / points
+	if interval < time.Second {
+		return time.Second
+	}
+	if roundUp {
+		return metric.CeilStandardInterval(interval)
+	}
+	return metric.FloorStandardInterval(interval)
+}
+
+// QueryMetrics is the shared query policy boundary. Raw and aggregate slices are
+// transferred directly from batch primitives; nullable chart data is allocated only
+// when explicitly requested by Fill. All work uses ctx and one captured clock.
+func QueryMetrics(ctx context.Context, req MetricsQuery) (MetricsResult, error) {
+	if ctx == nil {
+		return MetricsResult{}, fmt.Errorf("%w: context is required", metric.ErrInvalidArgument)
+	}
+	if err := ctx.Err(); err != nil {
+		return MetricsResult{}, err
+	}
+	result := MetricsResult{Definitions: map[string]metric.Definition{}, Raw: map[string][]metric.Point{}, Values: map[string]map[metric.Aggregation][]metric.AggregatePoint{}, Intervals: map[string]time.Duration{}, Latest: map[string]map[string]metric.Point{}}
+	if req.Mode == "" {
+		req.Mode = QuerySeries
+	}
+	if req.Source == "" {
+		req.Source = SourceAuto
+	}
+	if req.Grouping == "" {
+		req.Grouping = GroupMerged
+	}
+	if req.Order == "" {
+		req.Order = metric.OrderAsc
+	}
+	if req.Rounding == "" {
+		req.Rounding = RoundFloor
+	}
+	if err := validateMetricsQuery(req); err != nil {
+		return result, err
+	}
+	now := req.Now
+	if now.IsZero() {
+		now = time.Now()
+	}
+	now = now.UTC()
+	req.Start, req.End = req.Start.UTC(), req.End.UTC()
+	req.Entities.IDs = uniqueQueryIDs(req.Entities.IDs)
+	// Never let the primitive's empty entity filter broaden an empty authorized set.
+
+	s := req.Store
+	if s == nil {
+		s = GetStore()
+	}
+	if s == nil {
+		return result, fmt.Errorf("metric store not enabled")
+	}
+	names := make([]string, 0, len(req.Specs))
+	for _, spec := range req.Specs {
+		names = append(names, spec.Name)
+	}
+	loadDefinitions := s.GetMetrics
+	if req.CompactDefinitions {
+		loadDefinitions = s.GetMetricsCompact
+	}
+	if !req.Entities.All && len(req.Entities.IDs) == 0 {
+		definitions, err := loadDefinitions(ctx, names)
+		if err != nil {
+			return result, err
+		}
+		result.Definitions = definitions
+		result.IsRaw = req.Source == SourceRaw || (req.Source == SourceAuto && (req.Mode == QueryPoints || UsesRawWindow(req.Start, req.End, now)))
+		if err := checkQueryDefinitions(names, definitions, req.AllowMissing); err != nil {
+			return result, err
+		}
+		return result, ctx.Err()
+	}
+	if req.Mode == QueryLatestBefore {
+		definitions, err := loadDefinitions(ctx, names)
+		if err != nil {
+			return result, err
+		}
+		result.Definitions = definitions
+		if err := checkQueryDefinitions(names, definitions, req.AllowMissing); err != nil {
+			return result, err
+		}
+		for _, name := range names {
+			if _, ok := definitions[name]; !ok {
+				continue
+			}
+			ids := req.Entities.IDs
+
+			for _, id := range ids {
+				if err := ctx.Err(); err != nil {
+					return result, err
+				}
+				point, found, err := s.LatestBefore(ctx, name, id, req.End)
+				if err != nil {
+					return result, err
+				}
+				if found {
+					if result.Latest[name] == nil {
+						result.Latest[name] = map[string]metric.Point{}
+					}
+					result.Latest[name][id] = point
+				}
+			}
+		}
+		return result, ctx.Err()
+	}
+	result.IsRaw = req.Source == SourceRaw || (req.Source == SourceAuto && (req.Mode == QueryPoints || UsesRawWindow(req.Start, req.End, now)))
+	if result.IsRaw {
+		definitions, err := loadDefinitions(ctx, names)
+		if err != nil {
+			return result, err
+		}
+		result.Definitions = definitions
+		if err := checkQueryDefinitions(names, definitions, req.AllowMissing); err != nil {
+			return result, err
+		}
+		known := make([]string, 0, len(names))
+		for _, name := range names {
+			if _, ok := definitions[name]; ok {
+				known = append(known, name)
+				result.Intervals[name] = 0
+			}
+		}
+		if len(known) > 0 {
+			result.Raw, err = s.QueryBatch(ctx, metric.BatchQuery{MetricNames: known, EntityIDs: req.Entities.IDs, Start: req.Start, End: req.End, Tags: req.Tags, Order: req.Order, OmitTags: req.OmitTags, OmitLabels: req.OmitLabels, IncludeSeriesIdentity: req.Fill.Gaps || req.Fill.PingNegativeAsNull})
+			if err != nil {
+				return result, err
+			}
+		}
+	} else {
+		entityCount := len(req.Entities.IDs)
+		if needsTotalEntityCount(req) {
+			if req.Entities.All {
+				count, err := s.CountEntities(ctx, names)
+				if err != nil {
+					return result, err
+				}
+				entityCount = count
+			}
+			// Explicit IDs are a conservative dictionary-count upper bound; never
+			// scan every metric/tier simply to plan a response budget.
+		}
+		if entityCount < 1 {
+			entityCount = 1
+		}
+		specs := make([]metric.BatchSeriesSpec, 0, len(req.Specs))
+		for _, spec := range req.Specs {
+			interval := spec.Interval
+			if interval == 0 {
+				points := spec.MaxPoints
+				if points == 0 {
+					points = req.Budget.PerSeries
+				}
+				if points == 0 {
+					points = 500
+				}
+				if points == 500 && spec.MaxPoints == 0 && req.Budget.PerSeries == 0 && req.Budget.Total == -1 {
+					points = -1
+				}
+				interval = time.Second
+				if points != -1 {
+					if req.Budget.Total > 0 {
+						budget := req.Budget.Total / entityCount
+						if budget < 1 {
+							budget = 1
+						}
+						if budget < points {
+							points = budget
+						}
+					}
+					interval = DownsampleInterval(req.End.Sub(req.Start), points, req.Rounding == RoundCeil)
+				}
+				interval = s.CompatibleSeriesInterval(req.Start, now, interval)
+				if interval <= 0 {
+					interval = time.Second
+				}
+			}
+			aggregations := spec.Aggregations
+			if len(aggregations) == 0 {
+				aggregations = []metric.Aggregation{metric.AggAvg}
+			}
+			specs = append(specs, metric.BatchSeriesSpec{MetricName: spec.Name, Aggregations: aggregations, Interval: interval, PreserveSeries: req.Grouping == GroupSeries, GroupByEntity: req.Grouping == GroupEntity})
+			result.Intervals[spec.Name] = interval
+		}
+		batch, err := s.SeriesBatch(ctx, metric.BatchSeriesQuery{Specs: specs, EntityIDs: req.Entities.IDs, Start: req.Start, End: req.End, Tags: req.Tags, Order: req.Order, OmitTags: req.OmitTags, CompactDefinitions: req.CompactDefinitions, IncludeSeriesIdentity: req.Fill.Gaps || req.Fill.PingNegativeAsNull}, now)
+		if err != nil {
+			return result, err
+		}
+		result.Definitions, result.Values = batch.Definitions, batch.Values
+		// SeriesBatch already loads metadata. Validate its result rather than issue a
+		// second metadata query; unknown metrics otherwise look like empty series.
+		if err := checkQueryDefinitions(names, batch.Definitions, req.AllowMissing); err != nil {
+			return result, err
+		}
+		for _, name := range names {
+			if _, ok := batch.Definitions[name]; !ok {
+				delete(result.Intervals, name)
+				delete(result.Values, name)
+			}
+		}
+	}
+	if req.Fill.Gaps || req.Fill.PingNegativeAsNull {
+		filled, err := buildQuerySeriesOwned(ctx, req, result, req.Projection == ProjectionChart)
+		if err != nil {
+			return result, err
+		}
+		result.Filled = filled
+		if req.Projection == ProjectionChart {
+			result.Raw, result.Values = nil, nil
+		}
+	}
+	return result, ctx.Err()
+}
+
+// needsTotalEntityCount excludes explicit intervals and unlimited specs: neither
+// consumes the total adaptive budget. Keep its precedence aligned with planning.
+func needsTotalEntityCount(req MetricsQuery) bool {
+	if req.Budget.Total <= 0 {
+		return false
+	}
+	for _, spec := range req.Specs {
+		if spec.Interval != 0 {
+			continue
+		}
+		points := spec.MaxPoints
+		if points == 0 {
+			points = req.Budget.PerSeries
+		}
+		if points != -1 {
+			return true
+		}
+	}
+	return false
+}
+
+func validateMetricsQuery(req MetricsQuery) error {
+	invalid := func(message string) error { return fmt.Errorf("%w: %s", metric.ErrInvalidArgument, message) }
+	if req.OmitTags && (req.Fill.Gaps || req.Fill.PingNegativeAsNull) {
+		return invalid("omitting tags with filled output is unsupported: series identity must be preserved")
+	}
+	if req.Mode == QueryLatestBefore && (req.OmitTags || req.OmitLabels) {
+		return invalid("latest-before output omissions are unsupported")
+	}
+	switch req.Mode {
+	case QueryPoints, QuerySeries, QueryLatestBefore:
+	default:
+		return invalid("unsupported query mode")
+	}
+	switch req.Source {
+	case SourceAuto, SourceRaw, SourceRollup:
+	default:
+		return invalid("unsupported query source")
+	}
+	switch req.Grouping {
+	case GroupMerged, GroupEntity, GroupSeries:
+	default:
+		return invalid("unsupported query grouping")
+	}
+	switch req.Projection {
+	case "", ProjectionFull:
+	case ProjectionChart:
+		if req.Mode == QueryLatestBefore || (!req.Fill.Gaps && !req.Fill.PingNegativeAsNull) {
+			return invalid("chart projection requires fill and does not support latest-before")
+		}
+	default:
+		return invalid("unsupported query projection")
+	}
+	switch req.Rounding {
+	case RoundFloor, RoundCeil:
+	default:
+		return invalid("unsupported interval rounding")
+	}
+	if req.Order != metric.OrderAsc && req.Order != metric.OrderDesc {
+		return invalid("unsupported query order")
+	}
+	if req.Entities.All && len(req.Entities.IDs) > 0 {
+		return invalid("all entities and explicit IDs are mutually exclusive")
+	}
+	for _, id := range req.Entities.IDs {
+		if strings.TrimSpace(id) == "" {
+			return invalid("entity ID cannot be empty")
+		}
+	}
+	if req.End.IsZero() {
+		return invalid("end time is required")
+	}
+	if req.Mode != QueryLatestBefore && (req.Start.IsZero() || req.End.Before(req.Start)) {
+		return invalid("valid start and end times are required")
+	}
+	if req.Mode == QueryLatestBefore && (len(req.Tags) > 0 || req.Source != SourceAuto) {
+		return invalid("latest-before requires auto source and does not support tag filters")
+	}
+	if req.Mode == QueryPoints && req.Source == SourceRollup {
+		return invalid("points mode requires raw or auto source")
+	}
+	if req.Mode == QueryLatestBefore && req.Entities.All {
+		return invalid("latest-before requires explicit entity IDs")
+	}
+	if req.Budget.Total < -1 || req.Budget.PerSeries < -1 {
+		return invalid("invalid point budget")
+	}
+	if len(req.Specs) == 0 {
+		return invalid("at least one metric specification is required")
+	}
+	seen := map[string]bool{}
+	for _, spec := range req.Specs {
+		if strings.TrimSpace(spec.Name) == "" || seen[spec.Name] {
+			return invalid("metric names must be nonempty and unique")
+		}
+		seen[spec.Name] = true
+		if spec.Interval < 0 || spec.MaxPoints < -1 {
+			return invalid("invalid metric interval or point budget")
+		}
+		for _, agg := range spec.Aggregations {
+			// Reuse primitive aggregation validation even when the selected source is raw.
+			if err := (metric.AggregateQuery{Query: metric.Query{MetricName: spec.Name, Start: req.End, End: req.End}, Aggregation: agg, Interval: time.Second}).Validate(); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+// UsesRawWindow selects the public adaptive raw window; its end boundary is strict.
+func UsesRawWindow(start, end, now time.Time) bool {
+	return end.Sub(start) <= 10*time.Minute && end.After(now.Add(-10*time.Minute))
+}
+
+func uniqueQueryIDs(ids []string) []string {
+	out := make([]string, 0, len(ids))
+	seen := map[string]bool{}
+	for _, id := range ids {
+		if !seen[id] {
+			out = append(out, id)
+			seen[id] = true
+		}
+	}
+	return out
+}
+func checkQueryDefinitions(names []string, definitions map[string]metric.Definition, allowMissing bool) error {
+	if !allowMissing {
+		for _, name := range names {
+			if _, ok := definitions[name]; !ok {
+				return fmt.Errorf("%w: metric %q", metric.ErrNotFound, name)
+			}
+		}
+	}
+	return nil
+}
+
+// querySeriesKey consumes storage's immutable, collision-free output identity.
+// Manually constructed points (and low-level queries without the opt-in) retain
+// the old canonical fallback. Neither grouping nor ordering uses a hash.
+func querySeriesKey(entity string, tags map[string]string, identity *metric.SeriesIdentity) string {
+	if key := identity.Key(); key != "" {
+		return key
+	}
+	return metric.NewSeriesIdentity(entity, tags).Key()
+}
+
+func buildQuerySeries(ctx context.Context, req MetricsQuery, result MetricsResult) ([]MetricSeries, error) {
+	return buildQuerySeriesOwned(ctx, req, result, false)
+}
+
+// detach is only valid for QueryMetrics-owned source maps. Release each source
+// array after conversion, before allocating gaps or processing the next metric.
+func buildQuerySeriesOwned(ctx context.Context, req MetricsQuery, result MetricsResult, detach bool) ([]MetricSeries, error) {
+	out := []MetricSeries{}
+	for _, spec := range req.Specs {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		if _, ok := result.Definitions[spec.Name]; !ok {
+			continue
+		}
+		aggregations := spec.Aggregations
+		if result.IsRaw {
+			aggregations = []metric.Aggregation{""}
+		} else if len(aggregations) == 0 {
+			aggregations = []metric.Aggregation{metric.AggAvg}
+		}
+		for aggIndex, agg := range aggregations {
+			if err := ctx.Err(); err != nil {
+				return nil, err
+			}
+			var seenEntities map[string]bool
+			if len(req.Entities.IDs) > 0 && (result.IsRaw || req.Grouping != GroupMerged) {
+				seenEntities = make(map[string]bool, len(req.Entities.IDs))
+			}
+			groups := map[string]*MetricSeries{}
+			add := func(entity string, tags map[string]string, identity *metric.SeriesIdentity, value MetricValue) {
+				// Exact samples always preserve identity; aggregate identity is determined by
+				// the requested primitive grouping and must not be collapsed again here.
+				key := querySeriesKey(entity, tags, identity)
+				series := groups[key]
+				if series == nil {
+					series = &MetricSeries{MetricName: spec.Name, EntityID: entity, Aggregation: agg, Tags: tags, Interval: result.Intervals[spec.Name]}
+					groups[key] = series
+					if seenEntities != nil {
+						seenEntities[entity] = true
+					}
+				}
+				if req.Fill.PingNegativeAsNull && (spec.Name == MetricPingLatency || spec.Name == MetricPingLoss) && value.Value != nil && *value.Value == -1 {
+					value.Value = nil
+				}
+				series.Points = append(series.Points, value)
+			}
+			if result.IsRaw {
+				for i, p := range result.Raw[spec.Name] {
+					if i%256 == 0 {
+						if err := ctx.Err(); err != nil {
+							return nil, err
+						}
+					}
+					v := p.Value
+					add(p.EntityID, p.Tags, p.SeriesIdentity, MetricValue{Time: p.Timestamp, Value: &v, Count: 1, Tags: p.Tags, Labels: p.Labels})
+				}
+			} else {
+				for i, p := range result.Values[spec.Name][agg] {
+					if i%256 == 0 {
+						if err := ctx.Err(); err != nil {
+							return nil, err
+						}
+					}
+					v := p.Value
+					add(p.EntityID, p.Tags, p.SeriesIdentity, MetricValue{Time: p.Bucket, Value: &v, Count: p.Count, Tags: p.Tags})
+				}
+			}
+			if detach {
+				if result.IsRaw {
+					delete(result.Raw, spec.Name)
+				} else {
+					// Duplicate aggregation specs historically produce duplicate series.
+					needed := false
+					for _, next := range aggregations[aggIndex+1:] {
+						needed = needed || next == agg
+					}
+					if !needed {
+						delete(result.Values[spec.Name], agg)
+					}
+				}
+			}
+			// Seed missing entities directly; do not scan all groups for each ID or
+			// allocate a placeholder point only to discard it.
+			seed := func(id string) {
+				groups[querySeriesKey(id, req.Tags, nil)] = &MetricSeries{MetricName: spec.Name, EntityID: id, Aggregation: agg, Tags: req.Tags, Interval: result.Intervals[spec.Name]}
+				if seenEntities != nil {
+					seenEntities[id] = true
+				}
+			}
+			if result.IsRaw || req.Grouping != GroupMerged {
+				for i, id := range req.Entities.IDs {
+					if i%256 == 0 {
+						if err := ctx.Err(); err != nil {
+							return nil, err
+						}
+					}
+					if !seenEntities[id] {
+						seed(id)
+					}
+				}
+			}
+			if len(groups) == 0 {
+				ids := req.Entities.IDs
+				if len(ids) == 0 || (!result.IsRaw && req.Grouping == GroupMerged) {
+					ids = []string{""}
+				}
+				for _, id := range ids {
+					seed(id)
+				}
+			}
+
+			keys := make([]string, 0, len(groups))
+			for key := range groups {
+				if len(keys)%256 == 0 {
+					if err := ctx.Err(); err != nil {
+						return nil, err
+					}
+				}
+				keys = append(keys, key)
+			}
+			if err := ctx.Err(); err != nil {
+				return nil, err
+			}
+			sort.Strings(keys)
+			for _, key := range keys {
+				if err := ctx.Err(); err != nil {
+					return nil, err
+				}
+				series := *groups[key]
+				sort.SliceStable(series.Points, func(i, j int) bool { return series.Points[i].Time.Before(series.Points[j].Time) })
+				if err := ctx.Err(); err != nil {
+					return nil, err
+				}
+				if req.Fill.Gaps {
+					var err error
+					series, err = fillMetricSeriesOwned(ctx, series, req.Start, req.End)
+					if err != nil {
+						return nil, err
+					}
+				}
+				if req.Order == metric.OrderDesc {
+					for i, j := 0, len(series.Points)-1; i < j; i, j = i+1, j-1 {
+						if i%256 == 0 {
+							if err := ctx.Err(); err != nil {
+								return nil, err
+							}
+						}
+						series.Points[i], series.Points[j] = series.Points[j], series.Points[i]
+					}
+				}
+				out = append(out, series)
+			}
+		}
+	}
+	return out, ctx.Err()
+}
+
+// FillMetricSeries matches the public chart's adaptive lower-quartile gap policy.
+func FillMetricSeries(series MetricSeries, start, end time.Time) MetricSeries {
+	filled, _ := fillMetricSeries(context.Background(), series, start, end)
+	return filled
+}
+
+func fillMetricSeries(ctx context.Context, series MetricSeries, start, end time.Time) (MetricSeries, error) {
+	if err := ctx.Err(); err != nil {
+		return series, err
+	}
+	// Public callers retain ownership, including spare capacity. Never write there.
+	series.Points = append([]MetricValue(nil), series.Points...)
+	return fillMetricSeriesOwned(ctx, series, start, end)
+}
+
+// fillMetricSeriesOwned inserts gaps backwards into exclusively owned storage.
+// Dense series reuse their point array; sparse series grow only when capacity
+// cannot accommodate the exact number of gaps, rather than always copying into
+// a second full point array. The public wrapper above remains non-mutating.
+func fillMetricSeriesOwned(ctx context.Context, series MetricSeries, start, end time.Time) (MetricSeries, error) {
+	if err := ctx.Err(); err != nil {
+		return series, err
+	}
+	deltas := make([]time.Duration, 0, len(series.Points))
+	for i := 1; i < len(series.Points); i++ {
+		if i%256 == 0 {
+			if err := ctx.Err(); err != nil {
+				return series, err
+			}
+		}
+		if d := series.Points[i].Time.Sub(series.Points[i-1].Time); d > 0 {
+			deltas = append(deltas, d)
+		}
+	}
+	if err := ctx.Err(); err != nil {
+		return series, err
+	}
+	expected := series.Interval
+	if len(deltas) >= 2 {
+		sort.Slice(deltas, func(i, j int) bool { return deltas[i] < deltas[j] })
+		if observed := deltas[(len(deltas)-1)/4]; observed > expected {
+			expected = observed
+		}
+	}
+	if err := ctx.Err(); err != nil {
+		return series, err
+	}
+	if expected > 0 {
+		series.Interval = expected
+	}
+	null := func(at time.Time) MetricValue { return MetricValue{Time: at.UTC(), Tags: series.Tags} }
+	n := len(series.Points)
+	if n == 0 {
+		series.Points = append(series.Points, null(start), null(end))
+		return series, ctx.Err()
+	}
+	leading := start.Before(series.Points[0].Time)
+	gap := func(i int) bool {
+		return i > 0 && expected > 0 && series.Points[i-1].Value != nil && series.Points[i].Value != nil && series.Points[i].Time.Sub(series.Points[i-1].Time) > expected+expected/2
+	}
+	extra := 0
+	if leading {
+		extra++
+	}
+	for i := 1; i < n; i++ {
+		if i%256 == 0 {
+			if err := ctx.Err(); err != nil {
+				return series, err
+			}
+		}
+		if gap(i) {
+			extra++
+		}
+	}
+	if extra == 0 {
+		return series, ctx.Err()
+	}
+	if cap(series.Points) < n+extra {
+		points := make([]MetricValue, n+extra)
+		copy(points, series.Points)
+		series.Points = points
+	} else {
+		series.Points = series.Points[:n+extra]
+	}
+	write := n + extra - 1
+	for i := n - 1; i >= 0; i-- {
+		if i%256 == 0 {
+			if err := ctx.Err(); err != nil {
+				return series, err
+			}
+		}
+		// Decide before moving: write may equal i, but never precedes it.
+		insert := gap(i)
+		var at time.Time
+		if insert {
+			at = series.Points[i-1].Time.Add(expected)
+		}
+		series.Points[write] = series.Points[i]
+		write--
+		if insert {
+			series.Points[write] = null(at)
+			write--
+		}
+	}
+	if leading {
+		series.Points[0] = null(start)
+	}
+	return series, ctx.Err()
+}

internal/metricstore/report_batcher.go

@@ -341,7 +341,7 @@ func writeReportBatch(ctx context.Context, reports []v2.Report) ([]v2.Report, er
 			state.mu.Unlock()
 		}
 		if !values.initialized {
-			totalUp, hasUp, err := s.LatestBefore(ctx, MetricNetTotalUp, report.UUID, report.UpdatedAt)
+			totalUp, hasUp, err := latestReportPoint(ctx, s, MetricNetTotalUp, report.UUID, report.UpdatedAt)
 			if err != nil {
 				logger.Errorf("metricstore", "failed to restore previous upload counter for %s: %v", report.UUID, err)
 			} else {
@@ -349,7 +349,7 @@ func writeReportBatch(ctx context.Context, reports []v2.Report) ([]v2.Report, er
 				values.totalUpAt = totalUp.Timestamp
 				values.hasUp = hasUp
 			}
-			totalDown, hasDown, err := s.LatestBefore(ctx, MetricNetTotalDown, report.UUID, report.UpdatedAt)
+			totalDown, hasDown, err := latestReportPoint(ctx, s, MetricNetTotalDown, report.UUID, report.UpdatedAt)
 			if err != nil {
 				logger.Errorf("metricstore", "failed to restore previous download counter for %s: %v", report.UUID, err)
 			} else {

internal/metricstore/report_mapping.go

@@ -49,8 +49,23 @@ func reportMetricPoints(report v2.Report, trafficUp, trafficDown int64) []metric
 	return points
 }
 
+// latestReportPoint preserves independent counter restoration errors: one failed
+// direction must not prevent the other from restoring its baseline.
+func latestReportPoint(ctx context.Context, s *metric.Store, metricName, entityID string, before time.Time) (metric.Point, bool, error) {
+	loaded, err := QueryMetrics(ctx, MetricsQuery{
+		Store: s, Mode: QueryLatestBefore, Source: SourceAuto,
+		Entities: EntityScope{IDs: []string{entityID}},
+		Specs:    []MetricSpec{{Name: metricName}}, End: before, AllowMissing: true,
+	})
+	if err != nil {
+		return metric.Point{}, false, err
+	}
+	point, ok := loaded.Latest[metricName][entityID]
+	return point, ok, nil
+}
+
 func latestReportCounter(ctx context.Context, s *metric.Store, metricName, entityID string, before time.Time) (int64, bool, error) {
-	point, ok, err := s.LatestBefore(ctx, metricName, entityID, before)
+	point, ok, err := latestReportPoint(ctx, s, metricName, entityID, before)
 	if err != nil {
 		return 0, false, err
 	}

internal/sampling/sampling.go

@@ -0,0 +1,147 @@
+// Package sampling provides deterministic proportional downsampling used by
+// history responses. It is independent of transport and storage.
+package sampling
+
+import (
+	"math"
+	"sort"
+)
+
+// Group is one named set of candidate points.
+type Group[K comparable] struct {
+	Key    K
+	Length int
+}
+
+type remainder struct {
+	idx  int
+	frac float64
+}
+
+// AllocateTargets splits maxTotal across groups proportionally.
+func AllocateTargets[K comparable](groups []Group[K], maxTotal int) map[K]int {
+	result := make(map[K]int, len(groups))
+	if maxTotal < 0 {
+		for _, g := range groups {
+			result[g.Key] = g.Length
+		}
+		return result
+	}
+	total := 0
+	for _, g := range groups {
+		total += g.Length
+	}
+	if total <= maxTotal {
+		for _, g := range groups {
+			result[g.Key] = g.Length
+		}
+		return result
+	}
+	remainders := make([]remainder, 0, len(groups))
+	sumTargets := 0
+	for i, g := range groups {
+		if g.Length <= 0 {
+			result[g.Key] = 0
+			continue
+		}
+		raw := float64(g.Length) * float64(maxTotal) / float64(total)
+		t := int(math.Floor(raw))
+		if t > g.Length {
+			t = g.Length
+		}
+		result[groups[i].Key] = t
+		sumTargets += t
+		remainders = append(remainders, remainder{i, raw - float64(t)})
+	}
+	if sumTargets < maxTotal {
+		need := maxTotal - sumTargets
+		sortRemainders(remainders, false)
+		for _, r := range remainders {
+			if need == 0 {
+				break
+			}
+			g := groups[r.idx]
+			if result[g.Key] < g.Length {
+				result[g.Key]++
+				need--
+			}
+		}
+		for need > 0 {
+			for _, g := range groups {
+				if need == 0 {
+					break
+				}
+				if result[g.Key] < g.Length {
+					result[g.Key]++
+					need--
+				}
+			}
+			if need > 0 {
+				break
+			}
+		}
+	} else if sumTargets > maxTotal {
+		over := sumTargets - maxTotal
+		sortRemainders(remainders, true)
+		for _, r := range remainders {
+			if over == 0 {
+				break
+			}
+			g := groups[r.idx]
+			if result[g.Key] > 0 {
+				result[g.Key]--
+				over--
+			}
+		}
+		for over > 0 {
+			for _, g := range groups {
+				if over == 0 {
+					break
+				}
+				if result[g.Key] > 0 {
+					result[g.Key]--
+					over--
+				}
+			}
+			if over > 0 {
+				break
+			}
+		}
+	}
+	return result
+}
+
+func sortRemainders(r []remainder, asc bool) {
+	sort.Slice(r, func(i, j int) bool {
+		if asc {
+			return r[i].frac < r[j].frac
+		}
+		return r[i].frac > r[j].frac
+	})
+}
+
+// SampleEvenly selects k points including both endpoints.
+func SampleEvenly[T any](in []T, k int) []T {
+	n := len(in)
+	if k <= 0 || n == 0 {
+		return []T{}
+	}
+	if k >= n {
+		return in
+	}
+	out := make([]T, 0, k)
+	if k == 1 {
+		out = append(out, in[n-1])
+		return out
+	}
+	for i := 0; i < k; i++ {
+		idx := int(math.Round(float64(i) * float64(n-1) / float64(k-1)))
+		if idx < 0 {
+			idx = 0
+		} else if idx >= n {
+			idx = n - 1
+		}
+		out = append(out, in[idx])
+	}
+	return out
+}

pkg/metric/batch_series.go

@@ -28,12 +28,13 @@ type seriesIdentity struct {
 }
 
 type seriesReadMeta struct {
-	id         int64
-	metricName string
-	entityID   string
-	tagsHash   string
-	tagsJSON   string
-	tags       map[string]string
+	outputIdentity *SeriesIdentity
+	id             int64
+	metricName     string
+	entityID       string
+	tagsHash       string
+	tagsJSON       string
+	tags           map[string]string
 }
 
 type rollupRateSample struct {
@@ -42,20 +43,23 @@ type rollupRateSample struct {
 }
 
 type rollupAggregateState struct {
-	summary     *queryRollupSummary
-	rateSamples []rollupRateSample
-	tags        map[string]string
+	outputIdentity *SeriesIdentity
+	summary        *queryRollupSummary
+	rateSamples    []rollupRateSample
+	tags           map[string]string
 }
 
 type metricSeriesAccumulator struct {
-	spec        BatchSeriesSpec
-	compression float64
-	needDigest  bool
-	needSummary bool
-	needRate    bool
-	groups      map[rollupKey]*rollupAggregateState
-	rateGroups  map[rollupKey]*rollupAggregateState
-	states      *queryStateArena
+	includeSeriesIdentity bool
+	entityIdentities      map[string]*SeriesIdentity
+	spec                  BatchSeriesSpec
+	compression           float64
+	needDigest            bool
+	needSummary           bool
+	needRate              bool
+	groups                map[rollupKey]*rollupAggregateState
+	rateGroups            map[rollupKey]*rollupAggregateState
+	states                *queryStateArena
 }
 
 // SeriesBatch is the single rollup query implementation. It resolves metric
@@ -82,16 +86,36 @@ func (s *Store) GetMetrics(ctx context.Context, metricNames []string) (map[strin
 	return s.loadBatchDefinitions(ctx, metricNames)
 }
 
+// GetMetricsCompact loads only execution metadata: Name, Type, and RetentionDays.
+// GetMetrics continues to return complete definitions.
+func (s *Store) GetMetricsCompact(ctx context.Context, metricNames []string) (map[string]Definition, error) {
+	if err := s.ensureOpen(); err != nil {
+		return nil, err
+	}
+	if len(metricNames) == 0 {
+		return map[string]Definition{}, nil
+	}
+	return s.loadBatchDefinitionsProjected(ctx, metricNames, true)
+}
+
 func (s *Store) seriesBatchAt(ctx context.Context, query BatchSeriesQuery, now time.Time, forcedResolutions map[string]time.Duration) (BatchSeriesResult, error) {
 	metricNames := make([]string, 0, len(query.Specs))
 	for _, spec := range query.Specs {
 		metricNames = append(metricNames, spec.MetricName)
 	}
-	definitions, err := s.loadBatchDefinitions(ctx, metricNames)
+	definitions, err := s.loadBatchDefinitionsProjected(ctx, metricNames, query.CompactDefinitions)
 	if err != nil {
 		return BatchSeriesResult{}, err
 	}
 
+	return s.seriesBatchDefined(ctx, query, now, forcedResolutions, definitions)
+}
+
+func (s *Store) seriesBatchDefined(ctx context.Context, query BatchSeriesQuery, now time.Time, forcedResolutions map[string]time.Duration, definitions map[string]Definition) (BatchSeriesResult, error) {
+	metricNames := make([]string, 0, len(query.Specs))
+	for _, spec := range query.Specs {
+		metricNames = append(metricNames, spec.MetricName)
+	}
 	result := BatchSeriesResult{
 		Definitions: definitions,
 		Values:      make(map[string]map[Aggregation][]AggregatePoint, len(query.Specs)),
@@ -104,7 +128,8 @@ func (s *Store) seriesBatchAt(ctx context.Context, query BatchSeriesQuery, now t
 		values := make(map[Aggregation][]AggregatePoint, len(spec.Aggregations))
 		accumulator := &metricSeriesAccumulator{
 			spec: spec, compression: s.cfg.RollupPolicy.compression(), states: &states,
-			groups: make(map[rollupKey]*rollupAggregateState), rateGroups: make(map[rollupKey]*rollupAggregateState),
+			includeSeriesIdentity: query.IncludeSeriesIdentity,
+			groups:                make(map[rollupKey]*rollupAggregateState), rateGroups: make(map[rollupKey]*rollupAggregateState),
 		}
 		for _, aggregation := range spec.Aggregations {
 			values[aggregation] = []AggregatePoint{}
@@ -130,9 +155,9 @@ func (s *Store) seriesBatchAt(ctx context.Context, query BatchSeriesQuery, now t
 		if len(policy.Tiers) == 0 {
 			continue
 		}
-		resolution := seriesResolutionForPolicy(query.Start, spec.Interval, now, policy)
-		if forced, ok := forcedResolutions[spec.MetricName]; ok {
-			resolution = forced
+		resolution, forced := forcedResolutions[spec.MetricName]
+		if !forced {
+			resolution = seriesResolutionForPolicy(query.Start, spec.Interval, now, policy)
 		}
 		key := batchSeriesGroupKey{resolution: resolution, needDigest: accumulator.needDigest}
 		group := groups[key]
@@ -144,11 +169,12 @@ func (s *Store) seriesBatchAt(ctx context.Context, query BatchSeriesQuery, now t
 		group.fields |= rollupFieldsForAggregations(spec.Aggregations)
 	}
 
-	seriesByID, seriesByIdentity, err := s.loadSeriesDictionary(ctx, seriesDictionaryPlan{
-		MetricNames: metricNames,
-		EntityIDs:   query.EntityIDs,
-		Tags:        query.Tags,
-	})
+	// Dictionary IDs can be created by a hot-to-persisted flush. Resolve them
+	// under the same view lock as both scans so no continuously present series
+	// disappears between dictionary lookup and backing-tier selection.
+	s.rollupViewMu.RLock()
+	defer s.rollupViewMu.RUnlock()
+	seriesByID, seriesByIdentity, err := s.loadSeriesDictionary(ctx, seriesDictionaryPlan{MetricNames: metricNames, EntityIDs: query.EntityIDs, Tags: query.Tags, OmitTags: query.OmitTags})
 	if err != nil {
 		return BatchSeriesResult{}, err
 	}
@@ -170,8 +196,6 @@ func (s *Store) seriesBatchAt(ctx context.Context, query BatchSeriesQuery, now t
 		return -1
 	})
 
-	s.rollupViewMu.RLock()
-	defer s.rollupViewMu.RUnlock()
 	for _, group := range orderedGroups {
 		if err := s.scanPersistedRollupGroup(ctx, query, group, seriesByID, accumulators); err != nil {
 			return BatchSeriesResult{}, err
@@ -192,15 +216,23 @@ func (s *Store) seriesBatchAt(ctx context.Context, query BatchSeriesQuery, now t
 }
 
 func (s *Store) loadBatchDefinitions(ctx context.Context, metricNames []string) (map[string]Definition, error) {
+	return s.loadBatchDefinitionsProjected(ctx, metricNames, false)
+}
+
+func (s *Store) loadBatchDefinitionsProjected(ctx context.Context, metricNames []string, compact bool) (map[string]Definition, error) {
 	args := make([]any, 0, len(metricNames))
 	placeholders := make([]string, 0, len(metricNames))
 	for _, metricName := range metricNames {
 		args = append(args, metricName)
 		placeholders = append(placeholders, s.dialect.placeholder(len(args)))
 	}
+	columns := "name, type, unit, description, retention_days, metadata, created_at_milli, updated_at_milli"
+	if compact {
+		columns = "name, type, retention_days"
+	}
 	rows, err := s.reader().QueryContext(ctx, fmt.Sprintf(
-		"SELECT name, type, unit, description, retention_days, metadata, created_at_milli, updated_at_milli FROM %s WHERE name IN (%s) ORDER BY name ASC",
-		s.tables.definitions, strings.Join(placeholders, ", "),
+		"SELECT %s FROM %s WHERE name IN (%s) ORDER BY name ASC",
+		columns, s.tables.definitions, strings.Join(placeholders, ", "),
 	), args...)
 	if err != nil {
 		return nil, err
@@ -208,7 +240,13 @@ func (s *Store) loadBatchDefinitions(ctx context.Context, metricNames []string)
 	defer rows.Close()
 	definitions := make(map[string]Definition, len(metricNames))
 	for rows.Next() {
-		definition, err := scanDefinition(rows)
+		var definition Definition
+		var err error
+		if compact {
+			err = rows.Scan(&definition.Name, &definition.Type, &definition.RetentionDays)
+		} else {
+			definition, err = scanDefinition(rows)
+		}
 		if err != nil {
 			return nil, err
 		}
@@ -229,16 +267,22 @@ func (s *Store) loadSeriesDictionary(ctx context.Context, plan seriesDictionaryP
 	for rows.Next() {
 		var meta seriesReadMeta
 		var rawTags any
-		if err := rows.Scan(&meta.id, &meta.metricName, &meta.entityID, &meta.tagsHash, &rawTags); err != nil {
+		targets := []any{&meta.id, &meta.metricName, &meta.entityID, &meta.tagsHash}
+		if !plan.OmitTags {
+			targets = append(targets, &rawTags)
+		}
+		if err := rows.Scan(targets...); err != nil {
 			return nil, nil, err
 		}
-		meta.tagsJSON, err = rawJSONToString(rawTags)
-		if err != nil {
-			return nil, nil, err
-		}
-		meta.tags, err = rollupTagsFromJSON(meta.tagsJSON)
-		if err != nil {
-			return nil, nil, err
+		if !plan.OmitTags {
+			meta.tagsJSON, err = rawJSONToString(rawTags)
+			if err != nil {
+				return nil, nil, err
+			}
+			meta.tags, err = rollupTagsFromJSON(meta.tagsJSON)
+			if err != nil {
+				return nil, nil, err
+			}
 		}
 		byID[meta.id] = &meta
 		byIdentity[seriesIdentity{metricName: meta.metricName, entityID: meta.entityID, tagsHash: meta.tagsHash}] = &meta
@@ -261,13 +305,40 @@ func (s *Store) scanPersistedRollupGroup(ctx context.Context, query BatchSeriesQ
 	if err != nil || !found {
 		return err
 	}
-	rendered := s.dialect.renderRollupRead(s.tables, s.cfg.TablePrefix+"rollups_resolution_bucket_idx", rollupReadPlan{
+	plan := rollupReadPlan{
 		SeriesIDs:    seriesIDs,
 		ResolutionID: resolutionID,
 		StartMilli:   bucketStartMillis(query.Start.UnixMilli(), group.key.resolution.Milliseconds()),
 		EndMilli:     query.End.UnixMilli(),
 		Fields:       group.fields,
+	}
+	indexName := plan.indexName(s.cfg.TablePrefix)
+	return forEachRollupReadPlan(s.cfg.Driver, plan, func(chunk rollupReadPlan) error {
+		return s.scanPersistedRollupPlan(ctx, chunk, indexName, seriesByID, accumulators)
 	})
+}
+
+// MySQL prepared statements allow at most 65535 parameters. Leave headroom for
+// fixed predicates and future fields; other dialects bind the series as one array.
+const mysqlRollupSeriesChunkSize = 60000
+
+func forEachRollupReadPlan(driver Driver, plan rollupReadPlan, consume func(rollupReadPlan) error) error {
+	chunkSize := len(plan.SeriesIDs)
+	if driver == DriverMySQL && chunkSize > mysqlRollupSeriesChunkSize {
+		chunkSize = mysqlRollupSeriesChunkSize
+	}
+	for start := 0; start < len(plan.SeriesIDs); start += chunkSize {
+		chunk := plan
+		chunk.SeriesIDs = plan.SeriesIDs[start:min(start+chunkSize, len(plan.SeriesIDs))]
+		if err := consume(chunk); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (s *Store) scanPersistedRollupPlan(ctx context.Context, plan rollupReadPlan, indexName string, seriesByID map[int64]*seriesReadMeta, accumulators map[string]*metricSeriesAccumulator) error {
+	rendered := s.dialect.renderRollupRead(s.tables, indexName, plan)
 	rows, err := s.reader().QueryContext(ctx, rendered.Query, rendered.Args...)
 	if err != nil {
 		return err
@@ -277,25 +348,25 @@ func (s *Store) scanPersistedRollupGroup(ctx context.Context, query BatchSeriesQ
 	var sum, sumSq, min, max, firstVal, lastVal float64
 	var digestBlob []byte
 	destinations := []any{&seriesID, &bucket, &count}
-	if group.fields&rollupReadSum != 0 {
+	if plan.Fields&rollupReadSum != 0 {
 		destinations = append(destinations, &sum)
 	}
-	if group.fields&rollupReadSumSq != 0 {
+	if plan.Fields&rollupReadSumSq != 0 {
 		destinations = append(destinations, &sumSq)
 	}
-	if group.fields&rollupReadMin != 0 {
+	if plan.Fields&rollupReadMin != 0 {
 		destinations = append(destinations, &min)
 	}
-	if group.fields&rollupReadMax != 0 {
+	if plan.Fields&rollupReadMax != 0 {
 		destinations = append(destinations, &max)
 	}
-	if group.fields&rollupReadFirst != 0 {
+	if plan.Fields&rollupReadFirst != 0 {
 		destinations = append(destinations, &firstVal, &firstTS)
 	}
-	if group.fields&rollupReadLast != 0 {
+	if plan.Fields&rollupReadLast != 0 {
 		destinations = append(destinations, &lastVal, &lastTS)
 	}
-	if group.fields&rollupReadDigest != 0 {
+	if plan.Fields&rollupReadDigest != 0 {
 		destinations = append(destinations, &digestBlob)
 	}
 	for rows.Next() {
@@ -374,7 +445,7 @@ func (s *Store) scanHotRollupGroup(query BatchSeriesQuery, group *batchSeriesGro
 				continue
 			}
 		}
-		meta, matched, err := memorySeriesMeta(seriesByIdentity, key.metricName, key.entityID, key.tagsHash, bucket.tagsJSON, query.Tags)
+		meta, matched, err := memorySeriesMeta(seriesByIdentity, key.metricName, key.entityID, key.tagsHash, bucket.tagsJSON, query.Tags, query.OmitTags)
 		if err != nil {
 			return err
 		}
@@ -413,7 +484,7 @@ func (s *Store) scanCoarseRollupGroup(query BatchSeriesQuery, group *batchSeries
 		}
 		sortRollupKeys(childKeys)
 		firstChild := parent.children[childKeys[0]]
-		meta, matched, err := memorySeriesMeta(seriesByIdentity, key.metricName, key.entityID, key.tagsHash, firstChild.tagsJSON, query.Tags)
+		meta, matched, err := memorySeriesMeta(seriesByIdentity, key.metricName, key.entityID, key.tagsHash, firstChild.tagsJSON, query.Tags, query.OmitTags)
 		if err != nil {
 			return err
 		}
@@ -430,20 +501,27 @@ func (s *Store) scanCoarseRollupGroup(query BatchSeriesQuery, group *batchSeries
 	return nil
 }
 
-func memorySeriesMeta(seriesByIdentity map[seriesIdentity]*seriesReadMeta, metricName, entityID, tagsHash, tagsJSON string, filter map[string]string) (*seriesReadMeta, bool, error) {
+func memorySeriesMeta(seriesByIdentity map[seriesIdentity]*seriesReadMeta, metricName, entityID, tagsHash, tagsJSON string, filter map[string]string, omitTags bool) (*seriesReadMeta, bool, error) {
 	identity := seriesIdentity{metricName: metricName, entityID: entityID, tagsHash: tagsHash}
 	if meta := seriesByIdentity[identity]; meta != nil {
 		return meta, true, nil
 	}
-	tags, err := rollupTagsFromJSON(tagsJSON)
-	if err != nil {
-		return nil, false, err
+	var tags map[string]string
+	if !omitTags || len(filter) > 0 {
+		var err error
+		tags, err = rollupTagsFromJSON(tagsJSON)
+		if err != nil {
+			return nil, false, err
+		}
 	}
 	for key, value := range filter {
 		if tags[key] != value {
 			return nil, false, nil
 		}
 	}
+	if omitTags {
+		tags = nil
+	}
 	meta := &seriesReadMeta{metricName: metricName, entityID: entityID, tagsHash: tagsHash, tagsJSON: tagsJSON, tags: tags}
 	seriesByIdentity[identity] = meta
 	return meta, true, nil
@@ -459,8 +537,10 @@ func stringSet(values []string) map[string]struct{} {
 
 func (a *metricSeriesAccumulator) consume(meta *seriesReadMeta, sourceBucket, count int64, sum, sumSq, min, max, firstVal float64, firstTS int64, lastVal float64, lastTS int64, digest *TDigest) *rollupAggregateState {
 	key := rollupKey{bucket: bucketStartMillis(sourceBucket, a.spec.Interval.Milliseconds())}
-	if a.spec.PreserveSeries {
+	if a.spec.PreserveSeries || a.spec.GroupByEntity {
 		key.entityID = meta.entityID
+	}
+	if a.spec.PreserveSeries {
 		key.tagsHash = meta.tagsHash
 	}
 	if a.needSummary {
@@ -470,6 +550,7 @@ func (a *metricSeriesAccumulator) consume(meta *seriesReadMeta, sourceBucket, co
 			if a.spec.PreserveSeries {
 				state.tags = meta.tags
 			}
+			state.outputIdentity = a.outputIdentity(meta, key.entityID, a.spec.PreserveSeries)
 			a.groups[key] = state
 		}
 		state.summary.merge(count, sum, sumSq, min, max, firstVal, firstTS, lastVal, lastTS, digest, a.needDigest)
@@ -487,7 +568,7 @@ func (a *metricSeriesAccumulator) consume(meta *seriesReadMeta, sourceBucket, co
 		if state == nil {
 			// Rate-only state is smaller than a summary slot; do not reserve an
 			// unused summary/digest for every rate output bucket.
-			state = &rollupAggregateState{tags: meta.tags}
+			state = &rollupAggregateState{tags: meta.tags, outputIdentity: a.outputIdentity(meta, meta.entityID, true)}
 			a.rateGroups[rateKey] = state
 		}
 		state.rateSamples = append(state.rateSamples, rollupRateSample{timestamp: lastTS, value: lastVal})
@@ -495,6 +576,30 @@ func (a *metricSeriesAccumulator) consume(meta *seriesReadMeta, sourceBucket, co
 	return a.groups[key]
 }
 
+// outputIdentity reuses the dictionary series snapshot across all buckets and
+// aggregations. Merged/entity output has nil Tags and is cached separately;
+// using tagsHash here would wrongly split omitted Tags and risk hash collisions.
+func (a *metricSeriesAccumulator) outputIdentity(meta *seriesReadMeta, entity string, preserve bool) *SeriesIdentity {
+	if !a.includeSeriesIdentity {
+		return nil
+	}
+	if preserve {
+		if meta.outputIdentity == nil {
+			meta.outputIdentity = NewSeriesIdentity(entity, meta.tags)
+		}
+		return meta.outputIdentity
+	}
+	if a.entityIdentities == nil {
+		a.entityIdentities = make(map[string]*SeriesIdentity)
+	}
+	identity := a.entityIdentities[entity]
+	if identity == nil {
+		identity = NewSeriesIdentity(entity, nil)
+		a.entityIdentities[entity] = identity
+	}
+	return identity
+}
+
 func mergeEncodedRollupDigest(target *TDigest, count int64, min, max float64, blob []byte) error {
 	if count == 0 {
 		return nil
@@ -590,12 +695,13 @@ func (a *metricSeriesAccumulator) points(order Order) (map[Aggregation][]Aggrega
 				count = int(state.summary.count)
 			}
 			points = append(points, AggregatePoint{
-				MetricName: a.spec.MetricName,
-				EntityID:   key.entityID,
-				Bucket:     fromMillis(key.bucket),
-				Value:      value,
-				Count:      count,
-				Tags:       state.tags,
+				SeriesIdentity: state.outputIdentity,
+				MetricName:     a.spec.MetricName,
+				EntityID:       key.entityID,
+				Bucket:         fromMillis(key.bucket),
+				Value:          value,
+				Count:          count,
+				Tags:           state.tags,
 			})
 		}
 		values[aggregation] = points

pkg/metric/migrations.go

@@ -93,6 +93,7 @@ func normalizedIndexesFor(prefix string, tables tables) []normalizedIndex {
 	return []normalizedIndex{
 		{name: prefix + "series_metric_entity_idx", table: tables.series, columns: "metric_name, entity_id"},
 		{name: prefix + "rollups_resolution_bucket_idx", table: tables.rollups, columns: "resolution_id, bucket_milli"},
+		{name: prefix + "rollups_series_resolution_bucket_idx", table: tables.rollups, columns: "series_id, resolution_id, bucket_milli"},
 	}
 }
 

pkg/metric/raw_points.go

@@ -330,14 +330,29 @@ func (s *Store) queryRawPointsBatch(ctx context.Context, query BatchQuery) (map[
 				continue
 			}
 		}
-		tags, ok, err := matchRawTags(series.tagsJSON, query.Tags)
+		var tags map[string]string
+		ok := true
+		var err error
+		if !query.OmitTags || len(query.Tags) > 0 {
+			tags, ok, err = matchRawTags(series.tagsJSON, query.Tags)
+		}
+		if query.OmitTags {
+			tags = nil
+		}
 		if err != nil {
 			return nil, err
 		}
 		if !ok {
 			continue
 		}
-		labels := make(map[uint32]map[string]string)
+		var identity *SeriesIdentity
+		if query.IncludeSeriesIdentity {
+			identity = NewSeriesIdentity(key.entityID, tags)
+		}
+		var labels map[uint32]map[string]string
+		if !query.OmitLabels {
+			labels = make(map[uint32]map[string]string)
+		}
 		directStart := sort.Search(len(series.samples), func(i int) bool { return series.samples[i].timestamp >= start })
 		directEnd := sort.Search(len(series.samples), func(i int) bool { return series.samples[i].timestamp > end })
 		additional := directEnd - directStart
@@ -359,7 +374,7 @@ func (s *Store) queryRawPointsBatch(ctx context.Context, query BatchQuery) (map[
 				if err != nil {
 					return nil, err
 				}
-				appendRawQueryPoint(metricPoints, key, tags, labelMap, sample)
+				appendRawQueryPoint(metricPoints, key, tags, labelMap, sample, identity)
 			}
 		}
 		for _, sample := range series.samples[directStart:directEnd] {
@@ -367,7 +382,7 @@ func (s *Store) queryRawPointsBatch(ctx context.Context, query BatchQuery) (map[
 			if err != nil {
 				return nil, err
 			}
-			appendRawQueryPoint(metricPoints, key, tags, labelMap, sample)
+			appendRawQueryPoint(metricPoints, key, tags, labelMap, sample, identity)
 		}
 	}
 	result := make(map[string][]Point, len(matched))
@@ -379,6 +394,9 @@ func (s *Store) queryRawPointsBatch(ctx context.Context, query BatchQuery) (map[
 }
 
 func rawSampleLabels(series *rawSeries, labels map[uint32]map[string]string, labelID uint32) (map[string]string, error) {
+	if labels == nil {
+		return nil, nil
+	}
 	if labelMap := labels[labelID]; labelMap != nil {
 		return labelMap, nil
 	}
@@ -390,9 +408,10 @@ func rawSampleLabels(series *rawSeries, labels map[uint32]map[string]string, lab
 	return labelMap, nil
 }
 
-func appendRawQueryPoint(points *rawBatchPoints, key rawSeriesKey, tags, labels map[string]string, sample rawSample) {
+func appendRawQueryPoint(points *rawBatchPoints, key rawSeriesKey, tags, labels map[string]string, sample rawSample, identity *SeriesIdentity) {
 	points.points = append(points.points, Point{
-		MetricName: key.metricName, EntityID: key.entityID,
+		SeriesIdentity: identity,
+		MetricName:     key.metricName, EntityID: key.entityID,
 		Timestamp: fromMillis(sample.timestamp), Value: sample.value, Tags: tags, Labels: labels,
 	})
 	points.tagsHashes = append(points.tagsHashes, key.tagsHash)

pkg/metric/rollup_read_sql.go

@@ -10,6 +10,15 @@ type seriesDictionaryPlan struct {
 	MetricNames []string
 	EntityIDs   []string
 	Tags        map[string]string
+	OmitTags    bool
+}
+
+func (p seriesDictionaryPlan) columns() string {
+	columns := "s.id, s.metric_name, s.entity_id, s.tags_hash"
+	if !p.OmitTags {
+		columns += ", s.tags"
+	}
+	return columns
 }
 
 type rollupReadPlan struct {
@@ -20,6 +29,19 @@ type rollupReadPlan struct {
 	Fields       rollupReadFields
 }
 
+// Small resolved series sets should seek directly into each series, rather than
+// scan every series in the selected time range. Keep the time-first scan for
+// broad batches, where many separate seeks and the final sort cost more. This
+// bounded heuristic needs no extra cardinality query and also covers tag filters.
+const selectiveRollupSeriesLimit = 4
+
+func (p rollupReadPlan) indexName(prefix string) string {
+	if len(p.SeriesIDs) > 0 && len(p.SeriesIDs) <= selectiveRollupSeriesLimit {
+		return prefix + "rollups_series_resolution_bucket_idx"
+	}
+	return prefix + "rollups_resolution_bucket_idx"
+}
+
 type rollupReadFields uint16
 
 const (
@@ -57,7 +79,7 @@ func (d postgresDialect) renderSeriesDictionary(tables tables, _ string, plan se
 		parts = append(parts, d.jsonExtractEquals("s.tags", key, d.placeholder(len(args))))
 	}
 	return renderedSQL{
-		Query: fmt.Sprintf("SELECT s.id, s.metric_name, s.entity_id, s.tags_hash, s.tags FROM %s s WHERE %s ORDER BY s.id ASC", tables.series, strings.Join(parts, " AND ")),
+		Query: fmt.Sprintf("SELECT %s FROM %s s WHERE %s ORDER BY s.id ASC", plan.columns(), tables.series, strings.Join(parts, " AND ")),
 		Args:  args,
 	}
 }
@@ -75,7 +97,7 @@ func renderExpandedSeriesDictionary(d dialect, tables tables, indexHint string,
 		parts = append(parts, d.jsonExtractEquals("s.tags", key, d.placeholder(len(args))))
 	}
 	return renderedSQL{
-		Query: fmt.Sprintf("SELECT s.id, s.metric_name, s.entity_id, s.tags_hash, s.tags FROM %s s%s WHERE %s ORDER BY s.id ASC", tables.series, indexHint, strings.Join(parts, " AND ")),
+		Query: fmt.Sprintf("SELECT %s FROM %s s%s WHERE %s ORDER BY s.id ASC", plan.columns(), tables.series, indexHint, strings.Join(parts, " AND ")),
 		Args:  args,
 	}
 }

pkg/metric/series_identity.go

@@ -0,0 +1,29 @@
+package metric
+
+import "encoding/json"
+
+// SeriesIdentity is an immutable snapshot of output entity and Tags, not a
+// fingerprint. Its canonical key uses the historical chart JSON ordering,
+// including the distinction between nil and empty Tags. Labels are excluded.
+// Do not reuse an identity after changing a point's entity or Tags.
+type SeriesIdentity struct{ key string }
+
+// NewSeriesIdentity snapshots output identity once per series. It intentionally
+// uses output Tags (including nil when omitted), not the storage tags hash:
+// omission and merged aggregates must retain their historical chart grouping.
+func NewSeriesIdentity(entity string, tags map[string]string) *SeriesIdentity {
+	encoded, _ := json.Marshal(struct {
+		Entity string
+		Tags   map[string]string
+	}{entity, tags})
+	return &SeriesIdentity{key: string(encoded)}
+}
+
+// Key is collision-free with respect to the historical JSON grouping key and
+// preserves its lexical ordering. A nil/zero identity requests caller fallback.
+func (s *SeriesIdentity) Key() string {
+	if s == nil {
+		return ""
+	}
+	return s.key
+}

pkg/metric/store.go

@@ -1024,6 +1024,30 @@ func (s *Store) QueryBatch(ctx context.Context, query BatchQuery) (map[string][]
 	return s.queryRawPointsBatch(ctx, query.normalized())
 }
 
+// CountEntities returns the number of distinct entities registered for the
+// selected metrics. It reads the series dictionary, not a time range, so it is
+// a cheap upper bound used to divide an all-client point budget.
+func (s *Store) CountEntities(ctx context.Context, metricNames []string) (int, error) {
+	if err := s.ensureOpen(); err != nil {
+		return 0, err
+	}
+	if len(metricNames) == 0 {
+		return 0, nil
+	}
+	args := make([]any, 0, len(metricNames))
+	placeholders := make([]string, 0, len(metricNames))
+	for _, name := range metricNames {
+		args = append(args, name)
+		placeholders = append(placeholders, s.dialect.placeholder(len(args)))
+	}
+	var count int
+	err := s.reader().QueryRowContext(ctx, fmt.Sprintf(
+		"SELECT COUNT(DISTINCT entity_id) FROM %s WHERE metric_name IN (%s)",
+		s.tables.series, strings.Join(placeholders, ", "),
+	), args...).Scan(&count)
+	return count, err
+}
+
 // EntityIDs returns distinct entity ids that have exact in-memory samples or
 // persisted/hot rollups.
 func (s *Store) EntityIDs(ctx context.Context, query Query) ([]string, error) {

pkg/metric/types.go

@@ -194,6 +194,10 @@ func (d Definition) Validate() error {
 //
 // Point 表示某个实体在某一时刻的一条指标采样。
 type Point struct {
+	// SeriesIdentity is opt-in query metadata, never part of public JSON.
+	// Clear or rebuild it after changing EntityID or Tags (including map contents);
+	// grouping trusts the immutable snapshot. Compare Key(), not pointer addresses.
+	SeriesIdentity *SeriesIdentity `json:"-"`
 	// MetricName names the metric this sample belongs to.
 	//
 	// MetricName 表示该采样所属的指标名称。
@@ -294,12 +298,19 @@ type Query struct {
 // BatchQuery loads exact raw samples for multiple metrics and entities from one
 // shared time window.
 type BatchQuery struct {
-	MetricNames []string          `json:"metric_names"`
-	EntityIDs   []string          `json:"entity_ids,omitempty"`
-	Start       time.Time         `json:"start"`
-	End         time.Time         `json:"end"`
-	Tags        map[string]string `json:"tags,omitempty"`
-	Order       Order             `json:"order,omitempty"`
+	// IncludeSeriesIdentity snapshots output entity/Tags once per series.
+	// Default false keeps existing Go-value equality and JSON unchanged.
+	IncludeSeriesIdentity bool              `json:"-"`
+	MetricNames           []string          `json:"metric_names"`
+	EntityIDs             []string          `json:"entity_ids,omitempty"`
+	Start                 time.Time         `json:"start"`
+	End                   time.Time         `json:"end"`
+	Tags                  map[string]string `json:"tags,omitempty"`
+	Order                 Order             `json:"order,omitempty"`
+	// OmitTags skips output tag decoding, but does not change tag filters or ordering.
+	OmitTags bool `json:"omit_tags,omitempty"`
+	// OmitLabels skips decoding and returning sample labels.
+	OmitLabels bool `json:"omit_labels,omitempty"`
 }
 
 // BatchSeriesSpec describes the rollup output requested for one metric.
@@ -308,17 +319,29 @@ type BatchSeriesSpec struct {
 	Aggregations   []Aggregation
 	Interval       time.Duration
 	PreserveSeries bool
+	// GroupByEntity keeps entities separate while merging their tag variants.
+	// PreserveSeries takes precedence and additionally preserves tags. Rate
+	// output retains its existing per-series counter semantics.
+	GroupByEntity bool
 }
 
 // BatchSeriesQuery combines per-metric rollup specifications that share the
 // same time, entity, tag, and ordering filters.
 type BatchSeriesQuery struct {
-	Specs     []BatchSeriesSpec
-	EntityIDs []string
-	Start     time.Time
-	End       time.Time
-	Tags      map[string]string
-	Order     Order
+	// IncludeSeriesIdentity snapshots output entity/Tags once per series.
+	// Default false keeps existing Go-value equality and JSON unchanged.
+	IncludeSeriesIdentity bool `json:"-"`
+	Specs                 []BatchSeriesSpec
+	EntityIDs             []string
+	Start                 time.Time
+	End                   time.Time
+	Tags                  map[string]string
+	Order                 Order
+	// OmitTags skips output tags without changing filters or series grouping.
+	OmitTags bool
+	// CompactDefinitions returns only Name, Type, and RetentionDays, avoiding
+	// descriptive metadata loading and decoding. The default returns full definitions.
+	CompactDefinitions bool
 }
 
 // BatchSeriesResult contains definitions and aggregate values keyed by metric
@@ -504,6 +527,10 @@ func (q AggregateQuery) Validate() error {
 //
 // AggregatePoint 表示一个聚合桶的结果。
 type AggregatePoint struct {
+	// SeriesIdentity is opt-in query metadata, never part of public JSON.
+	// Clear or rebuild it after changing EntityID or Tags (including map contents);
+	// grouping trusts the immutable snapshot. Compare Key(), not pointer addresses.
+	SeriesIdentity *SeriesIdentity `json:"-"`
 	// MetricName is the metric represented by the bucket.
 	//
 	// MetricName 是该桶代表的指标名称。

web/rpc/jsonrpc/common.go

@@ -14,6 +14,7 @@ import (
 	"github.com/komari-monitor/komari/database/models"
 	"github.com/komari-monitor/komari/database/tasks"
 	"github.com/komari-monitor/komari/internal/config"
+	"github.com/komari-monitor/komari/internal/metricstore"
 	"github.com/komari-monitor/komari/pkg/rpc"
 	v2 "github.com/komari-monitor/komari/protocol/v2"
 	"github.com/komari-monitor/komari/utils"
@@ -60,7 +61,7 @@ func getPingStatsForNode(uuid string, pingTasks []models.PingTask) map[string]pi
 	}
 	end := time.Now().UTC()
 	start := end.Add(-1 * time.Hour)
-	recs, err := tasks.GetPingRecords(uuid, -1, start, end)
+	recs, err := metricstore.GetPingRecords(context.Background(), uuid, -1, start, end)
 	if err != nil || len(recs) == 0 {
 		empty := map[string]pingStat{}
 		pingStatsCache.Set(key, empty, cache.DefaultExpiration)

web/rpc/jsonrpc/common.record.go

@@ -2,7 +2,6 @@ package jsonrpc
 
 import (
 	"context"
-	"math"
 	"sort"
 	"time"
 
@@ -10,6 +9,8 @@ import (
 	"github.com/komari-monitor/komari/database/models"
 	recordsdb "github.com/komari-monitor/komari/database/records"
 	"github.com/komari-monitor/komari/database/tasks"
+	"github.com/komari-monitor/komari/internal/metricstore"
+	"github.com/komari-monitor/komari/internal/sampling"
 	"github.com/komari-monitor/komari/pkg/rpc"
 )
 
@@ -81,8 +82,19 @@ func getRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpc
 
 	switch params.Type {
 	case "load":
-		// fetch load records
-		recs, err := getLoadRecordsCombined(params.UUID, startTime, endTime)
+		// resolve maxCount before querying so the store can choose a coarser
+		// interval instead of materializing points that will be sampled away.
+		maxCount := params.MaxCount
+		if maxCount == 0 {
+			maxCount = 4000
+		}
+		// Fetch only the metrics backing load_type. Unknown types keep the
+		// compatible full-query path; the later filter still zeroes other fields.
+		opts := recordsdb.RecordQueryOptions{
+			MetricNames: recordsdb.LoadTypeMetricNames(params.LoadType),
+			MaxPoints:   maxCount,
+		}
+		recs, err := recordsdb.QueryRecords(ctx, params.UUID, startTime, endTime, opts)
 		if err != nil {
 			return nil, rpc.MakeError(rpc.InternalError, "Failed to fetch records", err.Error())
 		}
@@ -98,12 +110,6 @@ func getRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpc
 			recs = filtered
 		}
 
-		// resolve maxCount default for load
-		maxCount := params.MaxCount
-		if maxCount == 0 {
-			maxCount = 4000
-		}
-
 		// optional load_type filtering -> group by client
 		if params.LoadType != "" && params.LoadType != "all" {
 			items := filterRecordsByLoadType(recs, params.LoadType)
@@ -174,7 +180,7 @@ func getRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpc
 		if taskId == 0 {
 			taskId = -1
 		}
-		recs, err := tasks.GetPingRecords(params.UUID, taskId, startTime, endTime)
+		recs, err := metricstore.GetPingRecords(ctx, params.UUID, taskId, startTime, endTime)
 		if err != nil {
 			return nil, rpc.MakeError(rpc.InternalError, "Failed to fetch ping records", err.Error())
 		}
@@ -213,52 +219,32 @@ func getRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpc
 
 		response := &Resp{Count: 0, Records: []RecordsResp{}, From: startTime.UTC(), To: endTime.UTC()}
 
-		// stats per client
-		clientStats := make(map[string]struct {
-			total int
-			loss  int
-			min   int
-			max   int
-		})
-
-		for _, r := range recs {
-			rr := RecordsResp{
+		maxCount := params.MaxCount
+		if maxCount == 0 {
+			maxCount = 4000
+		}
+		displayRecs := recs
+		if maxCount != -1 && len(recs) > maxCount {
+			displayRecs = tasks.SamplePingRecordsByTask(recs, maxCount)
+		}
+		for _, r := range displayRecs {
+			response.Records = append(response.Records, RecordsResp{
 				TaskId: r.TaskId,
 				Time:   r.Time,
 				Value:  r.Value,
 				Client: r.Client,
-			}
-			st := clientStats[r.Client]
-			st.total++
-			if r.Value < 0 {
-				st.loss++
-			} else {
-				if st.min == 0 || r.Value < st.min {
-					st.min = r.Value
-				}
-				if r.Value > st.max {
-					st.max = r.Value
-				}
-			}
-			clientStats[r.Client] = st
-			response.Records = append(response.Records, rr)
+			})
 		}
 
-		if len(clientStats) > 0 {
-			response.BasicInfo = make([]ClientBasicInfo, 0, len(clientStats))
-			for client, st := range clientStats {
-				if client != "" && !isAdmin && hidden[client] {
-					continue
-				}
-				loss := float64(0)
-				if st.total > 0 {
-					loss = float64(st.loss) / float64(st.total) * 100
-				}
+		summary := tasks.SummarizePingRecords(recs)
+		if len(summary.Clients) > 0 {
+			response.BasicInfo = make([]ClientBasicInfo, 0, len(summary.Clients))
+			for client, st := range summary.Clients {
 				response.BasicInfo = append(response.BasicInfo, ClientBasicInfo{
 					Client: client,
-					Loss:   loss,
-					Min:    st.min,
-					Max:    st.max,
+					Loss:   st.LossPercent,
+					Min:    st.Min,
+					Max:    st.Max,
 				})
 			}
 		}
@@ -278,86 +264,10 @@ func getRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpc
 					continue
 				}
 			}
-			total := 0
-			lossCount := 0
-			minLat := 0
-			maxLat := 0
-			sum := 0
-			valid := 0
-			latestVal := -1
-			var latestTs time.Time
-			// 收集该任务的所有有效(非丢包)延迟值以计算百分位
-			latencies := make([]int, 0, 64)
-			for _, r := range recs {
-				if r.TaskId != t.Id {
-					continue
-				}
-				if params.UUID != "" && r.Client != params.UUID {
-					continue
-				}
-				total++
-				if r.Value < 0 { // 丢包
-					lossCount++
-					continue
-				}
-				valid++
-				sum += r.Value
-				latencies = append(latencies, r.Value)
-				if minLat == 0 || r.Value < minLat {
-					minLat = r.Value
-				}
-				if r.Value > maxLat {
-					maxLat = r.Value
-				}
-				// track latest non-negative value
-				ts := r.Time
-				if latestTs.IsZero() || ts.After(latestTs) {
-					latestTs = ts
-					latestVal = r.Value
-				}
-			}
-
-			// 计算 P50 / P99
-			p50 := 0
-			p99 := 0
-			if len(latencies) > 0 {
-				sort.Ints(latencies)
-				getPercentileInt := func(values []int, percentile float64) int {
-					if len(values) == 0 {
-						return 0
-					}
-					if percentile <= 0 {
-						return values[0]
-					}
-					if percentile >= 1 {
-						return values[len(values)-1]
-					}
-					pos := (float64(len(values) - 1)) * percentile
-					lo := int(math.Floor(pos))
-					hi := int(math.Ceil(pos))
-					if lo == hi {
-						return values[lo]
-					}
-					frac := pos - float64(lo)
-					v := float64(values[lo]) + (float64(values[hi])-float64(values[lo]))*frac
-					return int(math.Round(v))
-				}
-				p50 = getPercentileInt(latencies, 0.50)
-				p99 = getPercentileInt(latencies, 0.99)
-			}
-			ratio := 0.0
-			if p50 > 0 && p99 >= p50 {
-				jitterMs := float64(p99 - p50)
-				adjustedBase := math.Max(math.Min(float64(p50), 50.0), 10.0)
-				ratio = jitterMs / adjustedBase
-			}
-			lossRate := 0.0
-			if total > 0 {
-				lossRate = float64(lossCount) / float64(total) * 100
-			}
-			avg := 0
-			if valid > 0 {
-				avg = sum / valid
+			st := summary.Tasks[t.Id]
+			latest := st.Latest
+			if st.Total == 0 {
+				latest = -1
 			}
 			info := map[string]any{
 				"id":            t.Id,
@@ -365,15 +275,15 @@ func getRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpc
 				"type":          t.Type,
 				"interval":      t.Interval,
 				"default_on":    t.DefaultOn,
-				"loss":          lossRate,
-				"min":           minLat,
-				"max":           maxLat,
-				"avg":           avg,
-				"latest":        latestVal,
-				"total":         total,
-				"p50":           p50,
-				"p99":           p99,
-				"p99_p50_ratio": ratio,
+				"loss":          st.LossPercent,
+				"min":           st.Min,
+				"max":           st.Max,
+				"avg":           st.Avg,
+				"latest":        latest,
+				"total":         st.Total,
+				"p50":           st.P50,
+				"p99":           st.P99,
+				"p99_p50_ratio": st.P99P50Ratio,
 			}
 			if params.UUID == "" && taskId != -1 { // retain existing behavior of exposing clients only when filtering by task
 				info["clients"] = t.Clients
@@ -381,44 +291,6 @@ func getRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpc
 			toList = append(toList, info)
 		}
 		response.Tasks = toList
-		// apply maxCount for ping
-		maxCount := params.MaxCount
-		if maxCount == 0 {
-			maxCount = 4000
-		}
-		if maxCount != -1 && len(response.Records) > maxCount {
-			// group records by TaskId for proportional downsampling
-			taskGroups := make(map[uint][]RecordsResp)
-			for _, r := range response.Records {
-				taskGroups[r.TaskId] = append(taskGroups[r.TaskId], r)
-			}
-
-			// sort each group by time
-			for taskId := range taskGroups {
-				sort.Slice(taskGroups[taskId], func(i, j int) bool {
-					return taskGroups[taskId][i].Time.Before(taskGroups[taskId][j].Time)
-				})
-			}
-
-			groupsMeta := make([]allocationGroup[uint], 0, len(taskGroups))
-			for taskId, records := range taskGroups {
-				groupsMeta = append(groupsMeta, allocationGroup[uint]{
-					key:    taskId,
-					length: len(records),
-				})
-			}
-			targets := allocateTargets(groupsMeta, maxCount)
-
-			// downsample each task group
-			downsampledRecords := make([]RecordsResp, 0, maxCount)
-			for taskId, records := range taskGroups {
-				targetCount := targets[taskId]
-				sampled := sampleEvenly(records, targetCount)
-				downsampledRecords = append(downsampledRecords, sampled...)
-			}
-
-			response.Records = downsampledRecords
-		}
 		response.Count = len(response.Records)
 		// sort by time asc
 		sort.Slice(response.Records, func(i, j int) bool {
@@ -430,19 +302,6 @@ func getRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpc
 	}
 }
 
-// ---------- helpers for load records ----------
-
-// getLoadRecordsCombined fetches records for a client or all clients within a time range,
-// combining recent short-term table and long-term table with 15-min grouping for recent part.
-func getLoadRecordsCombined(uuid string, start, end time.Time) ([]models.Record, error) {
-	// prefer the existing function when uuid provided
-	if uuid != "" {
-		return recordsdb.GetRecordsByClientAndTime(uuid, start, end)
-	}
-	// 所有客户端：统一通过 records 包查询，启用 metric store 时自动走 metric store
-	return recordsdb.GetRecordsByTime(start, end)
-}
-
 // ---------- downsampling helpers ----------
 
 type allocationGroup[K comparable] struct {
@@ -452,213 +311,33 @@ type allocationGroup[K comparable] struct {
 
 // allocateTargets splits maxTotal across groups proportionally to their lengths.
 func allocateTargets[K comparable](groups []allocationGroup[K], maxTotal int) map[K]int {
-	result := make(map[K]int, len(groups))
-	if maxTotal < 0 {
-		for _, g := range groups {
-			result[g.key] = g.length
-		}
-		return result
-	}
-	total := 0
+	converted := make([]sampling.Group[K], 0, len(groups))
 	for _, g := range groups {
-		total += g.length
+		converted = append(converted, sampling.Group[K]{Key: g.key, Length: g.length})
 	}
-	if total <= maxTotal {
-		for _, g := range groups {
-			result[g.key] = g.length
-		}
-		return result
-	}
-	// initial allocation based on proportion
-	type rem struct {
-		idx  int
-		frac float64
-	}
-	remainders := make([]rem, 0, len(groups))
-	sumTargets := 0
-	for i, g := range groups {
-		if g.length <= 0 {
-			result[g.key] = 0
-			continue
-		}
-		raw := float64(g.length) * float64(maxTotal) / float64(total)
-		t := int(math.Floor(raw))
-		if t > g.length {
-			t = g.length
-		}
-		result[groups[i].key] = t
-		sumTargets += t
-		remainders = append(remainders, rem{i, raw - float64(t)})
-	}
-	// distribute remaining by largest fractional parts
-	if sumTargets < maxTotal {
-		need := maxTotal - sumTargets
-		sort.Slice(remainders, func(i, j int) bool { return remainders[i].frac > remainders[j].frac })
-		for _, r := range remainders {
-			if need == 0 {
-				break
-			}
-			g := groups[r.idx]
-			cur := result[g.key]
-			if cur < g.length {
-				result[g.key] = cur + 1
-				need--
-			}
-		}
-		// if still left, second pass round-robin
-		if need > 0 {
-			for need > 0 {
-				for _, g := range groups {
-					if need == 0 {
-						break
-					}
-					if result[g.key] < g.length {
-						result[g.key]++
-						need--
-					}
-				}
-				if need > 0 {
-					break
-				}
-			}
-		}
-	} else if sumTargets > maxTotal {
-		over := sumTargets - maxTotal
-		sort.Slice(remainders, func(i, j int) bool { return remainders[i].frac < remainders[j].frac })
-		for _, r := range remainders {
-			if over == 0 {
-				break
-			}
-			g := groups[r.idx]
-			if result[g.key] > 0 {
-				result[g.key]--
-				over--
-			}
-		}
-		if over > 0 {
-			for over > 0 {
-				for _, g := range groups {
-					if over == 0 {
-						break
-					}
-					if result[g.key] > 0 {
-						result[g.key]--
-						over--
-					}
-				}
-				if over > 0 {
-					break
-				}
-			}
-		}
-	}
-	return result
+	return sampling.AllocateTargets(converted, maxTotal)
 }
 
 func sampleEvenly[T any](in []T, k int) []T {
-	n := len(in)
-	if k <= 0 || n == 0 {
-		return []T{}
-	}
-	if k >= n {
-		return in
-	}
-	out := make([]T, 0, k)
-	if k == 1 {
-		out = append(out, in[n-1])
-		return out
-	}
-	for i := 0; i < k; i++ {
-		idx := int(math.Round(float64(i) * float64(n-1) / float64(k-1)))
-		if idx < 0 {
-			idx = 0
-		} else if idx >= n {
-			idx = n - 1
-		}
-		out = append(out, in[idx])
-	}
-	return out
+	return sampling.SampleEvenly(in, k)
 }
 
-// flatRecord is a projection used when load_type is specified.
+// flatRecord is the legacy load_type response shape. The selected fields come
+// from records.ProjectLoad so public/admin adapters share one switch.
 type flatRecord struct {
-	Client         string    `json:"client"`
-	Time           time.Time `json:"time"`
-	Cpu            *float32  `json:"cpu,omitempty"`
-	Gpu            *float32  `json:"gpu,omitempty"`
-	Ram            *int64    `json:"ram,omitempty"`
-	RamTotal       *int64    `json:"ram_total,omitempty"`
-	Swap           *int64    `json:"swap,omitempty"`
-	SwapTotal      *int64    `json:"swap_total,omitempty"`
-	Load           *float32  `json:"load,omitempty"`
-	Temp           *float32  `json:"temp,omitempty"`
-	Disk           *int64    `json:"disk,omitempty"`
-	DiskTotal      *int64    `json:"disk_total,omitempty"`
-	NetIn          *int64    `json:"net_in,omitempty"`
-	NetOut         *int64    `json:"net_out,omitempty"`
-	NetTotalUp     *int64    `json:"net_total_up,omitempty"`
-	NetTotalDown   *int64    `json:"net_total_down,omitempty"`
-	Process        *int      `json:"process,omitempty"`
-	Connections    *int      `json:"connections,omitempty"`
-	ConnectionsUdp *int      `json:"connections_udp,omitempty"`
-	Uptime         *int64    `json:"uptime,omitempty"`
+	Client string    `json:"client"`
+	Time   time.Time `json:"time"`
+	recordsdb.LoadProjection
 }
 
 func filterRecordsByLoadType(recs []models.Record, loadType string) []flatRecord {
 	out := make([]flatRecord, 0, len(recs))
 	for _, r := range recs {
-		fr := flatRecord{Client: r.Client, Time: r.Time}
-		switch loadType {
-		case "cpu":
-			v := r.Cpu
-			fr.Cpu = &v
-		case "gpu":
-			v := r.Gpu
-			fr.Gpu = &v
-		case "ram":
-			v := r.Ram
-			fr.Ram = &v
-			vt := r.RamTotal
-			fr.RamTotal = &vt
-		case "swap":
-			v := r.Swap
-			fr.Swap = &v
-			vt := r.SwapTotal
-			fr.SwapTotal = &vt
-		case "load":
-			v := r.Load
-			fr.Load = &v
-		case "temp":
-			v := r.Temp
-			fr.Temp = &v
-		case "disk":
-			v := r.Disk
-			fr.Disk = &v
-			vt := r.DiskTotal
-			fr.DiskTotal = &vt
-		case "network":
-			vi := r.NetIn
-			vo := r.NetOut
-			vtu := r.NetTotalUp
-			vtd := r.NetTotalDown
-			fr.NetIn = &vi
-			fr.NetOut = &vo
-			fr.NetTotalUp = &vtu
-			fr.NetTotalDown = &vtd
-		case "process":
-			v := r.Process
-			fr.Process = &v
-		case "connections":
-			v := r.Connections
-			fr.Connections = &v
-			vu := r.ConnectionsUdp
-			fr.ConnectionsUdp = &vu
-		default:
-			// unknown type: fallback to all fields as a full record would be returned elsewhere
-			v := r.Cpu
-			fr.Cpu = &v
-		}
-		out = append(out, fr)
+		out = append(out, flatRecord{
+			Client:         r.Client,
+			Time:           r.Time,
+			LoadProjection: recordsdb.ProjectLoad(r, loadType),
+		})
 	}
 	return out
 }

web/rpc/jsonrpc/ping_visibility.go

@@ -0,0 +1,15 @@
+package jsonrpc
+
+import "github.com/komari-monitor/komari/database/models"
+
+// visiblePingRecords filters before any public aggregation or sampling.
+func visiblePingRecords(records []models.PingRecord, hidden map[string]bool) []models.PingRecord {
+	visible := records[:0]
+	for _, record := range records {
+		if record.Client != "" && hidden[record.Client] {
+			continue
+		}
+		visible = append(visible, record)
+	}
+	return visible
+}

web/rpc/jsonrpc/public.go

@@ -11,6 +11,7 @@ import (
 	"github.com/komari-monitor/komari/database/models"
 	"github.com/komari-monitor/komari/database/records"
 	"github.com/komari-monitor/komari/database/tasks"
+	"github.com/komari-monitor/komari/internal/metricstore"
 	"github.com/komari-monitor/komari/pkg/rpc"
 	"github.com/komari-monitor/komari/utils"
 	agent_runtime "github.com/komari-monitor/komari/web/agent"
@@ -162,7 +163,9 @@ func publicGetRecordsByUUID(ctx context.Context, req *rpc.JsonRpcRequest) (any,
 		return nil, rpc.MakeError(rpc.InvalidParams, "Invalid load_type parameter", nil)
 	}
 	now := time.Now().UTC()
-	clientRecords, err := records.GetRecordsByClientAndTime(params.UUID, now.Add(-time.Duration(hoursInt)*time.Hour), now)
+	clientRecords, err := records.QueryRecords(ctx, params.UUID, now.Add(-time.Duration(hoursInt)*time.Hour), now, records.RecordQueryOptions{
+		MetricNames: records.LoadTypeMetricNames(params.LoadType),
+	})
 	if err != nil {
 		return nil, rpc.MakeError(rpc.InternalError, "Failed to fetch records: "+err.Error(), nil)
 	}
@@ -179,8 +182,11 @@ func publicGetRecordsByUUID(ctx context.Context, req *rpc.JsonRpcRequest) (any,
 		}
 	}
 	if params.LoadType == "" || params.LoadType == "all" || params.LoadType == "gpu" {
-		gpuRecords, err := records.GetGPURecordsByClientAndTime(params.UUID, now.Add(-time.Duration(hoursInt)*time.Hour), now)
-		if err == nil && len(gpuRecords) > 0 {
+		gpuRecords, err := records.QueryGPURecords(ctx, params.UUID, now.Add(-time.Duration(hoursInt)*time.Hour), now)
+		if err != nil {
+			return nil, rpc.MakeError(rpc.InternalError, "Failed to fetch GPU records: "+err.Error(), nil)
+		}
+		if len(gpuRecords) > 0 {
 			gpuDevices := make(map[string]any)
 			for _, record := range gpuRecords {
 				deviceKey := strconv.Itoa(record.DeviceIndex)
@@ -239,44 +245,45 @@ func filterPublicRecordsByLoadType(recs []models.Record, loadType string) []map[
 	out := make([]map[string]any, 0, len(recs))
 	for _, r := range recs {
 		record := map[string]any{"client": r.Client, "time": r.Time}
+		p := records.ProjectLoad(r, loadType)
 		switch loadType {
 		case "cpu":
-			record["cpu"] = r.Cpu
+			record["cpu"] = *p.Cpu
 		case "gpu":
-			record["gpu"] = r.Gpu
+			record["gpu"] = *p.Gpu
 		case "ram":
-			record["ram"] = r.Ram
-			record["ram_total"] = r.RamTotal
-			if r.RamTotal > 0 {
-				record["ram_percent"] = float32(r.Ram) / float32(r.RamTotal) * 100
+			record["ram"] = *p.Ram
+			record["ram_total"] = *p.RamTotal
+			if *p.RamTotal > 0 {
+				record["ram_percent"] = float32(*p.Ram) / float32(*p.RamTotal) * 100
 			}
 		case "swap":
-			record["swap"] = r.Swap
-			record["swap_total"] = r.SwapTotal
-			if r.SwapTotal > 0 {
-				record["swap_percent"] = float32(r.Swap) / float32(r.SwapTotal) * 100
+			record["swap"] = *p.Swap
+			record["swap_total"] = *p.SwapTotal
+			if *p.SwapTotal > 0 {
+				record["swap_percent"] = float32(*p.Swap) / float32(*p.SwapTotal) * 100
 			}
 		case "load":
-			record["load"] = r.Load
+			record["load"] = *p.Load
 		case "temp":
-			record["temp"] = r.Temp
+			record["temp"] = *p.Temp
 		case "disk":
-			record["disk"] = r.Disk
-			record["disk_total"] = r.DiskTotal
-			if r.DiskTotal > 0 {
-				record["disk_percent"] = float32(r.Disk) / float32(r.DiskTotal) * 100
+			record["disk"] = *p.Disk
+			record["disk_total"] = *p.DiskTotal
+			if *p.DiskTotal > 0 {
+				record["disk_percent"] = float32(*p.Disk) / float32(*p.DiskTotal) * 100
 			}
 		case "network":
-			record["net_in"] = r.NetIn
-			record["net_out"] = r.NetOut
-			record["net_total_up"] = r.NetTotalUp
-			record["net_total_down"] = r.NetTotalDown
+			record["net_in"] = *p.NetIn
+			record["net_out"] = *p.NetOut
+			record["net_total_up"] = *p.NetTotalUp
+			record["net_total_down"] = *p.NetTotalDown
 		case "process":
-			record["process"] = r.Process
+			record["process"] = *p.Process
 		case "connections":
-			record["connections"] = r.Connections
-			record["connections_udp"] = r.ConnectionsUdp
-			record["connections_tcp"] = r.Connections - r.ConnectionsUdp
+			record["connections"] = *p.Connections
+			record["connections_udp"] = *p.ConnectionsUdp
+			record["connections_tcp"] = *p.Connections - *p.ConnectionsUdp
 		}
 		out = append(out, record)
 	}
@@ -287,9 +294,10 @@ func filterPublicRecordsByLoadType(recs []models.Record, loadType string) []map[
 
 func publicGetPingRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpcError) {
 	var params struct {
-		UUID   string `json:"uuid"`
-		TaskID string `json:"task_id"`
-		Hours  string `json:"hours"`
+		UUID     string `json:"uuid"`
+		TaskID   string `json:"task_id"`
+		Hours    string `json:"hours"`
+		MaxCount int    `json:"max_count"`
 	}
 	req.BindParams(&params)
 	if params.UUID == "" && params.TaskID == "" {
@@ -350,46 +358,35 @@ func publicGetPingRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *r
 		}
 	}
 
-	recs, err := tasks.GetPingRecords(params.UUID, taskId, startTime, endTime)
+	recs, err := metricstore.GetPingRecords(ctx, params.UUID, taskId, startTime, endTime)
 	if err != nil {
 		return nil, rpc.MakeError(rpc.InternalError, "Failed to fetch ping records: "+err.Error(), nil)
 	}
 
-	clientStats := make(map[string]struct {
-		total, loss, min, max int
-	})
-	for _, r := range recs {
+	// Filter before sampling and statistics so hidden clients cannot affect
+	// public loss/latency summaries or consume the visible response budget.
+	if !isLogin {
+		recs = visiblePingRecords(recs, hiddenMap)
+	}
+	displayRecs := recs
+	if params.MaxCount > 0 && len(recs) > params.MaxCount {
+		displayRecs = tasks.SamplePingRecordsByTask(recs, params.MaxCount)
+	}
+	for _, r := range displayRecs {
 		if r.Client != "" && !isLogin && hiddenMap[r.Client] {
 			continue
 		}
-		rec := recordsResp{Time: r.Time.UTC(), Value: r.Value, Client: r.Client, TaskId: r.TaskId}
-		stats := clientStats[r.Client]
-		stats.total++
-		if r.Value < 0 {
-			stats.loss++
-		} else {
-			if stats.min == 0 || r.Value < stats.min {
-				stats.min = r.Value
-			}
-			if r.Value > stats.max {
-				stats.max = r.Value
-			}
-		}
-		clientStats[r.Client] = stats
-		response.Records = append(response.Records, rec)
+		response.Records = append(response.Records, recordsResp{Time: r.Time.UTC(), Value: r.Value, Client: r.Client, TaskId: r.TaskId})
 	}
 
-	if len(clientStats) > 0 {
-		response.BasicInfo = make([]clientBasicInfo, 0, len(clientStats))
-		for client, stats := range clientStats {
+	summary := tasks.SummarizePingRecords(recs)
+	if len(summary.Clients) > 0 {
+		response.BasicInfo = make([]clientBasicInfo, 0, len(summary.Clients))
+		for client, stats := range summary.Clients {
 			if client != "" && !isLogin && hiddenMap[client] {
 				continue
 			}
-			loss := float64(0)
-			if stats.total > 0 {
-				loss = float64(stats.loss) / float64(stats.total) * 100
-			}
-			response.BasicInfo = append(response.BasicInfo, clientBasicInfo{Client: client, Loss: loss, Min: stats.min, Max: stats.max})
+			response.BasicInfo = append(response.BasicInfo, clientBasicInfo{Client: client, Loss: stats.LossPercent, Min: stats.Min, Max: stats.Max})
 		}
 	}
 
@@ -406,40 +403,11 @@ func publicGetPingRecords(ctx context.Context, req *rpc.JsonRpcRequest) (any, *r
 			if params.UUID != "" && !t.AppliesToClient(params.UUID) {
 				continue
 			}
-			totalCount, lossCount, minLatency, maxLatency, sumLatency, validCount := 0, 0, 0, 0, 0, 0
-			for _, r := range recs {
-				if r.TaskId != t.Id {
-					continue
-				}
-				if params.UUID != "" && r.Client != params.UUID {
-					continue
-				}
-				totalCount++
-				if r.Value < 0 {
-					lossCount++
-				} else {
-					validCount++
-					sumLatency += r.Value
-					if minLatency == 0 || r.Value < minLatency {
-						minLatency = r.Value
-					}
-					if r.Value > maxLatency {
-						maxLatency = r.Value
-					}
-				}
-			}
-			lossRate := float64(0)
-			if totalCount > 0 {
-				lossRate = float64(lossCount) / float64(totalCount) * 100
-			}
-			avgLatency := 0
-			if validCount > 0 {
-				avgLatency = sumLatency / validCount
-			}
+			st := summary.Tasks[t.Id]
 			taskInfo := map[string]any{
 				"id": t.Id, "name": t.Name, "type": t.Type, "interval": t.Interval,
-				"default_on": t.DefaultOn, "loss": lossRate, "min": minLatency,
-				"max": maxLatency, "avg": avgLatency, "total": totalCount,
+				"default_on": t.DefaultOn, "loss": st.LossPercent, "min": st.Min,
+				"max": st.Max, "avg": st.Avg, "total": st.Total,
 			}
 			if params.UUID == "" && taskId != -1 {
 				taskInfo["clients"] = t.Clients

web/rpc/jsonrpc/public.metric.go

@@ -2,6 +2,7 @@ package jsonrpc
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"sort"
@@ -199,75 +200,39 @@ func publicQueryMetrics(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc
 	}
 
 	metricFillEmpty := resolveMetricFillEmpty(params)
-	useRaw := publicMetricUsesRawWindow(start, end, queryNow)
-	definitions := make(map[string]metric.Definition, len(metricKeys))
-	rawValues := make(map[string][]metric.Point)
-	rollupValues := make(map[string]map[metric.Aggregation][]metric.AggregatePoint)
-	if len(entityIDs) > 0 && useRaw {
-		var err error
-		definitions, err = store.GetMetrics(ctx, metricKeys)
-		if err != nil {
-			return nil, rpc.MakeError(rpc.InternalError, "Failed to query metric definitions: "+err.Error(), nil)
-		}
-		for _, spec := range loadSpecs {
-			if _, ok := definitions[spec.metricKey]; !ok {
-				return nil, rpc.MakeError(rpc.InvalidParams, "unknown metric key: "+spec.metricKey, nil)
-			}
-		}
-		rawValues, err = store.QueryBatch(ctx, metric.BatchQuery{
-			MetricNames: metricKeys,
-			EntityIDs:   entityIDs,
-			Start:       start,
-			End:         end,
-			Tags:        params.Tags,
-			Order:       metric.OrderAsc,
-		})
-		if err != nil {
-			return nil, rpc.MakeError(rpc.InvalidParams, "Failed to query metrics: "+err.Error(), nil)
-		}
-	} else if len(entityIDs) > 0 {
-		batchSpecs := make([]metric.BatchSeriesSpec, 0, len(loadSpecs))
-		for i := range loadSpecs {
-			loadSpecs[i].interval = metricDownsampleInterval(end.Sub(start), loadSpecs[i].maxPoints)
-			loadSpecs[i].interval = store.CompatibleSeriesInterval(start, queryNow, loadSpecs[i].interval)
-			batchSpecs = append(batchSpecs, metric.BatchSeriesSpec{
-				MetricName:     loadSpecs[i].metricKey,
-				Aggregations:   []metric.Aggregation{loadSpecs[i].algorithm},
-				Interval:       loadSpecs[i].interval,
-				PreserveSeries: true,
-			})
-		}
-		loaded, err := store.SeriesBatch(ctx, metric.BatchSeriesQuery{
-			Specs:     batchSpecs,
-			EntityIDs: entityIDs,
-			Start:     start,
-			End:       end,
-			Tags:      params.Tags,
-			Order:     metric.OrderAsc,
-		}, queryNow)
-		if err != nil {
-			return nil, rpc.MakeError(rpc.InvalidParams, "Failed to query metrics: "+err.Error(), nil)
-		}
-		definitions = loaded.Definitions
-		rollupValues = loaded.Values
-		for _, spec := range loadSpecs {
-			if _, ok := definitions[spec.metricKey]; !ok {
-				return nil, rpc.MakeError(rpc.InvalidParams, "unknown metric key: "+spec.metricKey, nil)
-			}
-		}
-	} else {
-		var err error
-		definitions, err = store.GetMetrics(ctx, metricKeys)
-		if err != nil {
-			return nil, rpc.MakeError(rpc.InternalError, "Failed to query metric definitions: "+err.Error(), nil)
-		}
-		for _, spec := range loadSpecs {
-			if _, ok := definitions[spec.metricKey]; !ok {
-				return nil, rpc.MakeError(rpc.InvalidParams, "unknown metric key: "+spec.metricKey, nil)
-			}
+	querySpecs := make([]metricstore.MetricSpec, 0, len(loadSpecs))
+	for _, spec := range loadSpecs {
+		querySpecs = append(querySpecs, metricstore.MetricSpec{Name: spec.metricKey, Aggregations: []metric.Aggregation{spec.algorithm}, MaxPoints: spec.maxPoints})
+	}
+	projection := metricstore.ProjectionFull
+	if metricFillEmpty {
+		// This branch consumes Filled only; do not retain a second source view.
+		projection = metricstore.ProjectionChart
+	}
+	loaded, err := metricstore.QueryMetrics(ctx, metricstore.MetricsQuery{
+		Projection: projection,
+		Store:      store, Mode: metricstore.QuerySeries, Source: metricstore.SourceAuto,
+		Entities: metricstore.EntityScope{IDs: entityIDs}, Specs: querySpecs,
+		Start: start, End: end, Tags: params.Tags, Order: metric.OrderAsc,
+		Grouping: metricstore.GroupSeries, Rounding: metricstore.RoundCeil,
+		Fill: metricstore.FillPolicy{Gaps: metricFillEmpty, PingNegativeAsNull: metricFillEmpty}, AllowMissing: true, Now: queryNow,
+	})
+	if err != nil {
+		return nil, metricQueryRPCError(err)
+	}
+	definitions, rawValues, rollupValues := loaded.Definitions, loaded.Raw, loaded.Values
+	useRaw := loaded.IsRaw
+	for i := range loadSpecs {
+		if _, ok := definitions[loadSpecs[i].metricKey]; !ok {
+			return nil, rpc.MakeError(rpc.InvalidParams, "unknown metric key: "+loadSpecs[i].metricKey, nil)
 		}
+		loadSpecs[i].interval = loaded.Intervals[loadSpecs[i].metricKey]
 	}
 
+	filledByMetric := make(map[string][]metricstore.MetricSeries)
+	for _, filled := range loaded.Filled {
+		filledByMetric[filled.MetricName] = append(filledByMetric[filled.MetricName], filled)
+	}
 	series := make([]publicMetricSeries, 0, len(metricKeys)*maxInt(1, len(entityIDs)))
 	for _, spec := range loadSpecs {
 		def := definitions[spec.metricKey]
@@ -281,29 +246,31 @@ func publicQueryMetrics(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc
 			MaxPoints:     spec.maxPoints,
 			Downsampled:   !useRaw,
 		}
-		if useRaw {
+		if !useRaw {
+			item.DownsampleAlgorithm = string(spec.algorithm)
+			item.IntervalSeconds = spec.interval.Seconds()
+		}
+		if !metricFillEmpty && useRaw {
 			points := rawValues[spec.metricKey]
 			item.Points = make([]publicMetricPoint, 0, len(points))
 			for _, point := range points {
 				item.Points = append(item.Points, publicMetricPoint{
 					entityID: point.EntityID,
 					Time:     point.Timestamp.UTC(),
-					Value:    publicRawMetricValue(point.MetricName, point.Value, metricFillEmpty),
+					Value:    publicMetricValue(point.Value),
 					Count:    1,
 					Tags:     point.Tags,
 					Labels:   point.Labels,
 				})
 			}
-		} else {
-			item.DownsampleAlgorithm = string(spec.algorithm)
-			item.IntervalSeconds = spec.interval.Seconds()
+		} else if !metricFillEmpty {
 			points := rollupValues[spec.metricKey][spec.algorithm]
 			item.Points = make([]publicMetricPoint, 0, len(points))
 			for _, point := range points {
 				item.Points = append(item.Points, publicMetricPoint{
 					entityID: point.EntityID,
 					Time:     point.Bucket.UTC(),
-					Value:    publicRawMetricValue(point.MetricName, point.Value, metricFillEmpty),
+					Value:    publicMetricValue(point.Value),
 					Count:    point.Count,
 					Tags:     point.Tags,
 				})
@@ -311,10 +278,20 @@ func publicQueryMetrics(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc
 		}
 
 		byEntity := make(map[string][]publicMetricSeries, len(entityIDs))
-		for _, split := range splitPublicMetricSeries(item) {
-			if split.EntityID != "" {
+		if metricFillEmpty {
+			for _, filled := range filledByMetric[spec.metricKey] {
+				if !useRaw && filled.Aggregation != spec.algorithm {
+					continue
+				}
+				split := publicMetricSeriesFromCore(item, filled)
 				byEntity[split.EntityID] = append(byEntity[split.EntityID], split)
 			}
+		} else {
+			for _, split := range splitPublicMetricSeries(item) {
+				if split.EntityID != "" {
+					byEntity[split.EntityID] = append(byEntity[split.EntityID], split)
+				}
+			}
 		}
 		for _, entityID := range entityIDs {
 			entitySeries := byEntity[entityID]
@@ -323,12 +300,12 @@ func publicQueryMetrics(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc
 				empty.EntityID = entityID
 				empty.Count = 0
 				empty.Points = nil
+				if metricFillEmpty {
+					empty = adaptiveFillPublicMetricSeries(empty, start, end)
+				}
 				entitySeries = []publicMetricSeries{empty}
 			}
 			for _, split := range entitySeries {
-				if metricFillEmpty {
-					split = adaptiveFillPublicMetricSeries(split, start, end)
-				}
 				series = append(series, split)
 			}
 		}
@@ -344,6 +321,15 @@ func publicQueryMetrics(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc
 	}, nil
 }
 
+// metricQueryRPCError keeps invalid requests separate from storage/cancellation failures.
+func metricQueryRPCError(err error) *rpc.JsonRpcError {
+	code := rpc.InternalError
+	if errors.Is(err, metric.ErrInvalidArgument) || errors.Is(err, metric.ErrNotFound) {
+		code = rpc.InvalidParams
+	}
+	return rpc.MakeError(code, "Failed to query metrics: "+err.Error(), nil)
+}
+
 type publicMetricPointResult struct {
 	points      []publicMetricPoint
 	downsampled bool
@@ -359,56 +345,39 @@ func loadPublicMetricPoints(
 	fillEmpty bool,
 	now time.Time,
 ) (publicMetricPointResult, error) {
-	if publicMetricUsesRawWindow(query.Start, query.End, now) {
-		points, err := store.Query(ctx, query)
-		if err != nil {
-			return publicMetricPointResult{}, err
-		}
-		result := publicMetricPointResult{points: make([]publicMetricPoint, 0, len(points))}
-		for _, point := range points {
-			result.points = append(result.points, publicMetricPoint{
-				entityID: point.EntityID,
-				Time:     point.Timestamp.UTC(),
-				Value:    publicRawMetricValue(point.MetricName, point.Value, fillEmpty),
-				Count:    1,
-				Tags:     point.Tags,
-				Labels:   point.Labels,
-			})
-		}
-		return result, nil
+	if maxPoints <= 0 {
+		maxPoints = defaultMetricQueryPoints
 	}
-
-	interval := metricDownsampleInterval(query.End.Sub(query.Start), maxPoints)
-	interval = store.CompatibleSeriesInterval(query.Start, now, interval)
-	points, err := store.Series(ctx, metric.AggregateQuery{
-		Query:          query,
-		Aggregation:    algorithm,
-		Interval:       interval,
-		PreserveSeries: true,
-	}, now)
+	loaded, err := metricstore.QueryMetrics(ctx, metricstore.MetricsQuery{
+		Store: store, Mode: metricstore.QuerySeries, Source: metricstore.SourceAuto,
+		Entities: metricstore.EntityScope{IDs: []string{query.EntityID}},
+		Specs:    []metricstore.MetricSpec{{Name: query.MetricName, Aggregations: []metric.Aggregation{algorithm}, MaxPoints: maxPoints}},
+		Start:    query.Start, End: query.End, Tags: query.Tags, Order: query.Order,
+		Grouping: metricstore.GroupSeries, Rounding: metricstore.RoundCeil,
+		Fill: metricstore.FillPolicy{PingNegativeAsNull: fillEmpty}, Now: now,
+	})
 	if err != nil {
 		return publicMetricPointResult{}, err
 	}
-	result := publicMetricPointResult{
-		points:      make([]publicMetricPoint, 0, len(points)),
-		downsampled: true,
-		interval:    interval,
-	}
-	for _, point := range points {
-		result.points = append(result.points, publicMetricPoint{
-			entityID: point.EntityID,
-			Time:     point.Bucket.UTC(),
-			Value:    publicRawMetricValue(point.MetricName, point.Value, fillEmpty),
-			Count:    point.Count,
-			Tags:     point.Tags,
-		})
+	result := publicMetricPointResult{points: []publicMetricPoint{}, downsampled: !loaded.IsRaw, interval: loaded.Intervals[query.MetricName]}
+	if fillEmpty {
+		for _, series := range loaded.Filled {
+			result.points = append(result.points, publicMetricSeriesFromCore(publicMetricSeries{}, series).Points...)
+		}
+	} else if loaded.IsRaw {
+		for _, point := range loaded.Raw[query.MetricName] {
+			result.points = append(result.points, publicMetricPoint{entityID: point.EntityID, Time: point.Timestamp.UTC(), Value: publicMetricValue(point.Value), Count: 1, Tags: point.Tags, Labels: point.Labels})
+		}
+	} else {
+		for _, point := range loaded.Values[query.MetricName][algorithm] {
+			result.points = append(result.points, publicMetricPoint{entityID: point.EntityID, Time: point.Bucket.UTC(), Value: publicMetricValue(point.Value), Count: point.Count, Tags: point.Tags})
+		}
 	}
 	return result, nil
 }
 
 func publicMetricUsesRawWindow(start, end, now time.Time) bool {
-	retention := metricstore.DefaultRollupRawRetention
-	return end.Sub(start) <= retention && end.After(now.UTC().Add(-retention))
+	return metricstore.UsesRawWindow(start, end, now)
 }
 
 func publicGetPingMetricStats(ctx context.Context, req *rpc.JsonRpcRequest) (any, *rpc.JsonRpcError) {
@@ -458,10 +427,7 @@ func publicGetPingMetricStats(ctx context.Context, req *rpc.JsonRpcRequest) (any
 		maxPoints = defaultMetricQueryPoints
 	}
 	now := time.Now().UTC()
-	interval := metricDownsampleInterval(end.Sub(start), maxPoints)
-	interval = store.CompatibleSeriesInterval(start, now, interval)
-
-	groupsByEntity, err := loadPublicPingMetricAggregateGroups(ctx, store, entityIDs, start, end, interval, now)
+	groupsByEntity, interval, err := loadPublicPingMetricAggregateGroups(ctx, store, entityIDs, start, end, maxPoints, now)
 	if err != nil {
 		return nil, rpc.MakeError(rpc.InternalError, "Failed to query ping stats: "+err.Error(), nil)
 	}
@@ -550,61 +516,23 @@ func splitPublicMetricSeries(base publicMetricSeries) []publicMetricSeries {
 // is inferred per metric/entity/tag series, so sparse periodic data does not
 // expand into hundreds of artificial empty buckets.
 func adaptiveFillPublicMetricSeries(series publicMetricSeries, start, end time.Time) publicMetricSeries {
-	pointTimes := make([]time.Time, len(series.Points))
-	deltas := make([]time.Duration, 0, len(series.Points))
-	for i, point := range series.Points {
-		pointTimes[i] = point.Time
-		if i > 0 {
-			delta := point.Time.Sub(pointTimes[i-1])
-			if delta > 0 {
-				deltas = append(deltas, delta)
-			}
-		}
+	core := metricstore.MetricSeries{MetricName: series.MetricKey, EntityID: series.EntityID, Tags: series.Tags, Interval: time.Duration(series.IntervalSeconds * float64(time.Second))}
+	for _, point := range series.Points {
+		core.Points = append(core.Points, metricstore.MetricValue{Time: point.Time, Value: point.Value, Count: point.Count, Tags: point.Tags, Labels: point.Labels})
 	}
+	return publicMetricSeriesFromCore(series, metricstore.FillMetricSeries(core, start, end))
+}
 
-	expectedInterval := time.Duration(series.IntervalSeconds * float64(time.Second))
-	// Two deltas are the minimum needed to distinguish a regular cadence from
-	// one isolated long gap. A lower quartile keeps outages from inflating the
-	// inferred cadence when the rest of the series is regular.
-	if len(deltas) >= 2 {
-		sort.Slice(deltas, func(i, j int) bool { return deltas[i] < deltas[j] })
-		observedInterval := deltas[(len(deltas)-1)/4]
-		if observedInterval > expectedInterval {
-			expectedInterval = observedInterval
-		}
+func publicMetricSeriesFromCore(base publicMetricSeries, series metricstore.MetricSeries) publicMetricSeries {
+	base.EntityID = series.EntityID
+	base.Tags = series.Tags
+	base.IntervalSeconds = series.Interval.Seconds()
+	base.Points = make([]publicMetricPoint, 0, len(series.Points))
+	for _, point := range series.Points {
+		base.Points = append(base.Points, publicMetricPoint{entityID: series.EntityID, Time: point.Time.UTC(), Value: point.Value, Count: point.Count, Tags: point.Tags, Labels: point.Labels})
 	}
-	if expectedInterval > 0 {
-		series.IntervalSeconds = expectedInterval.Seconds()
-	}
-
-	nullPoint := func(at time.Time) publicMetricPoint {
-		return publicMetricPoint{
-			Time:  at.UTC(),
-			Value: nil,
-			Tags:  series.Tags,
-		}
-	}
-	filled := make([]publicMetricPoint, 0, len(series.Points)+2)
-	if len(pointTimes) == 0 || start.Before(pointTimes[0]) {
-		filled = append(filled, nullPoint(start))
-	}
-	for i, point := range series.Points {
-		if i > 0 && expectedInterval > 0 && series.Points[i-1].Value != nil && point.Value != nil {
-			delta := pointTimes[i].Sub(pointTimes[i-1])
-			if delta > expectedInterval+expectedInterval/2 {
-				filled = append(filled, nullPoint(pointTimes[i-1].Add(expectedInterval)))
-			}
-		}
-		filled = append(filled, point)
-	}
-	// Do not append a trailing null after real data. It would make the final chart
-	// bucket blank regardless of whether the tail is a collection delay or a gap.
-	if len(pointTimes) == 0 {
-		filled = append(filled, nullPoint(end))
-	}
-	series.Points = filled
-	series.Count = len(filled)
-	return series
+	base.Count = len(base.Points)
+	return base
 }
 
 func publicMetricTagsKey(tags map[string]string) string {
@@ -752,20 +680,6 @@ func publicMetricValue(value float64) *float64 {
 	return &value
 }
 
-func publicRawMetricValue(metricName string, value float64, fillEmpty bool) *float64 {
-	if isNullPingMetricValue(metricName, value, fillEmpty) {
-		return nil
-	}
-	return publicMetricValue(value)
-}
-
-func isNullPingMetricValue(metricName string, value float64, fillEmpty bool) bool {
-	if !fillEmpty || value != -1 {
-		return false
-	}
-	return metricName == metricstore.MetricPingLatency || metricName == metricstore.MetricPingLoss
-}
-
 type publicPingMetricAggregateGroups struct {
 	Avg           map[string][]metric.AggregatePoint
 	Min           map[string][]metric.AggregatePoint
@@ -778,7 +692,7 @@ type publicPingMetricAggregateGroups struct {
 	LossAvailable bool
 }
 
-func loadPublicPingMetricAggregateGroups(ctx context.Context, store *metric.Store, entityIDs []string, start, end time.Time, interval time.Duration, now time.Time) (map[string]publicPingMetricAggregateGroups, error) {
+func loadPublicPingMetricAggregateGroups(ctx context.Context, store *metric.Store, entityIDs []string, start, end time.Time, maxPoints int, now time.Time) (map[string]publicPingMetricAggregateGroups, time.Duration, error) {
 	latencyAggregations := []metric.Aggregation{
 		metric.AggAvg,
 		metric.AggMin,
@@ -788,18 +702,17 @@ func loadPublicPingMetricAggregateGroups(ctx context.Context, store *metric.Stor
 		metric.AggP99,
 		metric.AggStdDev,
 	}
-	loaded, err := store.SeriesBatch(ctx, metric.BatchSeriesQuery{
-		Specs: []metric.BatchSeriesSpec{
-			{MetricName: metricstore.MetricPingLatency, Aggregations: latencyAggregations, Interval: interval, PreserveSeries: true},
-			{MetricName: metricstore.MetricPingLoss, Aggregations: []metric.Aggregation{metric.AggAvg}, Interval: interval, PreserveSeries: true},
+	loaded, err := metricstore.QueryMetrics(ctx, metricstore.MetricsQuery{
+		Store: store, Mode: metricstore.QuerySeries, Source: metricstore.SourceRollup,
+		Specs: []metricstore.MetricSpec{
+			{Name: metricstore.MetricPingLatency, Aggregations: latencyAggregations, MaxPoints: maxPoints},
+			{Name: metricstore.MetricPingLoss, Aggregations: []metric.Aggregation{metric.AggAvg}, MaxPoints: maxPoints},
 		},
-		EntityIDs: entityIDs,
-		Start:     start,
-		End:       end,
-		Order:     metric.OrderAsc,
-	}, now)
+		Entities: metricstore.EntityScope{IDs: entityIDs}, Start: start, End: end, Order: metric.OrderAsc,
+		Grouping: metricstore.GroupSeries, Rounding: metricstore.RoundCeil, AllowMissing: true, Now: now,
+	})
 	if err != nil {
-		return nil, err
+		return nil, 0, err
 	}
 	latency := loaded.Values[metricstore.MetricPingLatency]
 	lossPoints := loaded.Values[metricstore.MetricPingLoss][metric.AggAvg]
@@ -834,7 +747,7 @@ func loadPublicPingMetricAggregateGroups(ctx context.Context, store *metric.Stor
 			LossAvailable: pingMetricGroupsHaveData(entityLoss),
 		}
 	}
-	return result, nil
+	return result, loaded.Intervals[metricstore.MetricPingLatency], nil
 }
 
 func groupPingMetricAggregatePointsByEntity(points []metric.AggregatePoint) map[string]map[string][]metric.AggregatePoint {
@@ -1099,21 +1012,6 @@ func normalizeMetricAggregation(raw string) string {
 	}
 }
 
-func metricDownsampleInterval(rangeDuration time.Duration, maxPoints int) time.Duration {
-	if maxPoints <= 0 {
-		maxPoints = defaultMetricQueryPoints
-	}
-	nanos := rangeDuration.Nanoseconds()
-	if nanos <= 0 {
-		return time.Second
-	}
-	interval := time.Duration((nanos + int64(maxPoints) - 1) / int64(maxPoints))
-	if interval < time.Second {
-		return time.Second
-	}
-	return metric.CeilStandardInterval(interval)
-}
-
 func maxInt(a, b int) int {
 	if a > b {
 		return a

web/rpc/jsonrpc/public_metric_test.go

@@ -207,29 +207,6 @@ func TestAdaptiveFillPublicMetricSeriesNeverEndsWithNullAfterData(t *testing.T)
 	}
 }
 
-func TestPublicPingMetricFillEmptyMapsMinusOneToNull(t *testing.T) {
-	for _, metricName := range []string{metricstore.MetricPingLatency, metricstore.MetricPingLoss} {
-		if value := publicRawMetricValue(metricName, -1, true); value != nil {
-			t.Fatalf("raw %s -1 should become null when fill_empty is enabled, got %v", metricName, *value)
-		}
-	}
-	if value := publicRawMetricValue(metricstore.MetricPingLatency, -1, true); value != nil {
-		t.Fatalf("downsampled ping -1 should become null when fill_empty is enabled, got %v", *value)
-	}
-}
-
-func TestPublicPingMetricMinusOneIsPreservedWithoutFillEmpty(t *testing.T) {
-	value := publicRawMetricValue(metricstore.MetricPingLatency, -1, false)
-	if value == nil || *value != -1 {
-		t.Fatalf("raw ping -1 should be preserved when fill_empty is disabled, got %v", value)
-	}
-
-	nonPing := publicRawMetricValue("temperature", -1, true)
-	if nonPing == nil || *nonPing != -1 {
-		t.Fatalf("negative values from non-ping metrics must be preserved, got %v", nonPing)
-	}
-}
-
 func TestPublicPingStatsFromAggregateGroupsUsesTaskNamesAndLossMetric(t *testing.T) {
 	base := time.Date(2026, 6, 18, 0, 0, 0, 0, time.UTC)
 	taskMap := map[string]models.PingTask{
@@ -310,17 +287,17 @@ func TestPublicPingMetricStatsIncludesZeroVolatility(t *testing.T) {
 }
 
 func TestMetricDownsampleIntervalCeilsToStandardInterval(t *testing.T) {
-	got := metricDownsampleInterval(30*24*time.Hour, 500)
+	got := metricstore.DownsampleInterval(30*24*time.Hour, 500, true)
 	if got != 2*time.Hour {
 		t.Fatalf("30d/500 should ceil to 2h, got %s", got)
 	}
 
-	got = metricDownsampleInterval(time.Hour, 500)
+	got = metricstore.DownsampleInterval(time.Hour, 500, true)
 	if got != 10*time.Second {
 		t.Fatalf("1h/500 should ceil to 10s, got %s", got)
 	}
 
-	got = metricDownsampleInterval(1000*24*time.Hour, 10)
+	got = metricstore.DownsampleInterval(1000*24*time.Hour, 10, true)
 	if got != 100*24*time.Hour {
 		t.Fatalf("ranges beyond the standard table should ceil to whole days, got %s", got)
 	}