Каталог разрезов и измеренный род агрегации

- род метрики выводится сверкой минутного слоя с часовым: часовое значение сходится с суммой минутных — накопительная, со средним — мгновенная, иначе `unknown` и свёртка не предлагается вовсе. На живом архиве (123 доставки, 31 метрика) 7 накопительных, 9 мгновенных, противоречащих часов ноль - `GET /api/v1/metrics` под токеном чтения отдаёт единицы, слои с границами и род вместе с основанием измерения; род нигде не хранится — он функция витрины, а витрина функция журнала, устаревать в нём нечему - миграция 00009: покрывающий индекс, чтобы каталог отвечал по учётным колонкам, не разжимая содержимое объектов
2026-08-02 19:23:59 +03:00
parent 98e0772ec5
commit 03edf1087d
39 changed files with 4744 additions and 58 deletions
@@ -0,0 +1,292 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+	"time"
+)
+
+// Запросы каталога вынесены константами: по ним же проверяется план выполнения.
+// Обе выборки обязаны отвечать по покрывающему индексу `bucket_catalog`, не
+// касаясь строк таблицы — вместе со строкой пришло бы и сжатое содержимое.
+const (
+	layerRangesQuery = `
+		SELECT metric, layer, units, min(first_ts), max(last_ts), sum(points)
+		FROM bucket
+		GROUP BY metric, layer, units
+		ORDER BY metric, layer, units`
+
+	// Общие часы: самые свежие часы, за которые у метрики есть объекты обоих
+	// слоёв. Возвращаются вместе с числом точек и единицами каждой стороны —
+	// эти колонки лежат в том же покрывающем индексе, а стоят они того, чтобы
+	// решать пригодность часа НЕ разжимая содержимое.
+	//
+	// Горизонт (`f.hour_utc <= ?`) обязателен. Час объекта берётся из метки в
+	// теле доставки, а тело мы не контролируем: без верхней границы одна
+	// доставка с метками в будущем занимает всё окно и подменяет измеренный род
+	// метрики. Свидетельство из будущего свидетельством не является.
+	commonHoursQuery = `
+		SELECT f.hour_utc, f.points, f.units, c.points, c.units
+		FROM bucket f
+		JOIN bucket c ON c.metric = ? AND c.layer = ? AND c.hour_utc = f.hour_utc
+		WHERE f.metric = ? AND f.layer = ? AND f.hour_utc <= ?
+		ORDER BY f.hour_utc DESC
+		LIMIT ?`
+)
+
+// hourPairsQuery строит выборку объектов окна: ОДИН запрос на любое число
+// часов. Отдельной функцией потому, что число подстановок переменное, а форма
+// «один запрос независимо от размера окна» — проверяемое свойство, а не
+// намерение.
+func hourPairsQuery(hours int) string {
+	return `
+		SELECT layer, hour_utc, payload FROM bucket
+		WHERE metric = ? AND layer IN (?, ?) AND hour_utc IN (` +
+		strings.TrimSuffix(strings.Repeat("?,", hours), ",") + `)`
+}
+
+// LayerRange — один разрез метрики: слой, единицы, границы данных, число точек.
+//
+// Границы — это границы ДАННЫХ, а не обещание покрытия: внутри диапазона законно
+// есть дыры (часы без доставок, периоды, чьи верхние слои не пережили
+// пересборку). Единицы входят в разрез потому, что группировка идёт вместе с
+// ними: расхождение единиц у объектов одной метрики обязано быть видно строкой,
+// а не выбираться молча.
+type LayerRange struct {
+	Metric string
+	Layer  string
+	Units  string
+	From   time.Time
+	To     time.Time
+	Points int
+}
+
+// HourPair — час, за который у метрики есть объекты обоих слоёв сразу.
+// Именно на таких часах и держится измерение рода агрегации.
+//
+// Точки заполнены не всегда: содержимое читается только у часов, прошедших
+// предварительный отбор по учётным колонкам (см. CatalogWindow). У остальных
+// известны число точек и единицы — этого хватает, чтобы признать час
+// непригодным, не разжимая ни байта.
+type HourPair struct {
+	Hour         time.Time
+	FinePoints   int
+	FineUnits    string
+	Fine         []Point
+	CoarsePoints int
+	CoarseUnits  string
+	Coarse       []Point
+}
+
+// CatalogWindow — что считать окном измерения. Все числа задаёт вызывающий:
+// правило измерения принадлежит домену, хранилище лишь выбирает по нему строки.
+type CatalogWindow struct {
+	// Fine и Coarse — мелкий и крупный слои сверки.
+	Fine, Coarse string
+	// Hours — сколько самых свежих общих часов брать.
+	Hours int
+	// Horizon — верхняя граница: часы позже неё в окно не входят.
+	Horizon time.Time
+	// CoarsePoints — сколько точек обязан нести объект крупного слоя, чтобы час
+	// стоило читать. MinFinePoints — сколько минимум обязан нести мелкий.
+	// Отбор ПРЕДВАРИТЕЛЬНЫЙ и строго слабее правила вердикта: он экономит
+	// разжатие заведомо непригодных часов, а решение принимает домен.
+	CoarsePoints  int
+	MinFinePoints int
+}
+
+// CatalogSnapshot — весь вход каталога, снятый ОДНОЙ транзакцией чтения.
+//
+// Единый снимок здесь не аккуратность. Приём идёт непрерывно, и фоновая свёртка
+// пишет в витрину во время запроса: разрезы, снятые до её коммита, и объекты
+// окна, прочитанные после, дали бы ответ, внутренне противоречивый и
+// неотличимый от обычного свежего. Тот же довод записан у отпечатка витрины, и
+// второй его экземпляр разошёлся бы с первым молча.
+type CatalogSnapshot struct {
+	// Layers — разрезы всех метрик, упорядоченные по метрике, слою и единицам.
+	Layers []LayerRange
+	// Pairs — по метрике её общие часы, от САМЫХ СВЕЖИХ к старым. Метрики, у
+	// которой нет объектов обоих слоёв, в карте нет вовсе.
+	Pairs map[string][]HourPair
+}
+
+// ReadCatalog снимает вход каталога: разрезы всех метрик и объекты окна.
+//
+// Число обращений к базе — `1 + 2×метрик` и от размера окна НЕ зависит: объекты
+// окна читаются пакетом, одним запросом на метрику. Чтение по объекту за раз
+// давало бы под сотню обращений на метрику, каждое своей транзакцией.
+//
+// Содержимое разжимается только у часов, прошедших отбор по учётным колонкам.
+// Разжимать всё подряд означало бы платить памятью за часы, чей вердикт заранее
+// известен: замер на раздутой витрине давал 109 МиБ аллокаций при нуле
+// пригодных часов.
+func (s *Store) ReadCatalog(ctx context.Context, w CatalogWindow) (CatalogSnapshot, error) {
+	out := CatalogSnapshot{Pairs: make(map[string][]HourPair)}
+	switch {
+	case w.Hours <= 0:
+		return CatalogSnapshot{}, fmt.Errorf("окно каталога: часов должно быть больше нуля, задано %d", w.Hours)
+	case w.Fine == w.Coarse:
+		return CatalogSnapshot{}, fmt.Errorf("окно каталога: слои сверки совпадают (%q)", w.Fine)
+	}
+
+	tx, err := s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
+	if err != nil {
+		return CatalogSnapshot{}, fmt.Errorf("begin read tx: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	out.Layers, err = readLayerRanges(ctx, tx)
+	if err != nil {
+		return CatalogSnapshot{}, err
+	}
+
+	for _, metric := range metricsWithBothLayers(out.Layers, w.Fine, w.Coarse) {
+		pairs, err := commonHours(ctx, tx, metric, w)
+		if err != nil {
+			return CatalogSnapshot{}, err
+		}
+		if len(pairs) == 0 {
+			continue
+		}
+		if err := readHourPairs(ctx, tx, metric, w, pairs); err != nil {
+			return CatalogSnapshot{}, err
+		}
+		out.Pairs[metric] = pairs
+	}
+	return out, nil
+}
+
+// readLayerRanges отвечает по покрывающему индексу: содержимое объектов ради
+// границ и счётчиков не разжимается и даже не читается.
+func readLayerRanges(ctx context.Context, tx *sql.Tx) ([]LayerRange, error) {
+	rows, err := tx.QueryContext(ctx, layerRangesQuery)
+	if err != nil {
+		return nil, fmt.Errorf("select layer ranges: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	out := make([]LayerRange, 0, 64)
+	for rows.Next() {
+		var r LayerRange
+		var from, to string
+		if err := rows.Scan(&r.Metric, &r.Layer, &r.Units, &from, &to, &r.Points); err != nil {
+			return nil, fmt.Errorf("scan layer range: %w", err)
+		}
+		if r.From, err = ParseTime(from); err != nil {
+			return nil, err
+		}
+		if r.To, err = ParseTime(to); err != nil {
+			return nil, err
+		}
+		out = append(out, r)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("select layer ranges: %w", err)
+	}
+	return out, nil
+}
+
+// metricsWithBothLayers отбирает метрики, у которых есть оба слоя. Без отбора
+// пришлось бы спрашивать общие часы у метрик, где второго слоя заведомо нет, —
+// два лишних запроса на каждую.
+func metricsWithBothLayers(layers []LayerRange, fine, coarse string) []string {
+	seen := make(map[string]map[string]bool, len(layers))
+	order := make([]string, 0, len(layers))
+	for _, l := range layers {
+		set, ok := seen[l.Metric]
+		if !ok {
+			set = map[string]bool{}
+			seen[l.Metric] = set
+			order = append(order, l.Metric)
+		}
+		set[l.Layer] = true
+	}
+
+	out := make([]string, 0, len(order))
+	for _, m := range order {
+		if seen[m][fine] && seen[m][coarse] {
+			out = append(out, m)
+		}
+	}
+	return out
+}
+
+// commonHours возвращает самые свежие часы окна, от свежих к старым, вместе с
+// учётными колонками обеих сторон. Содержимого не читает.
+func commonHours(ctx context.Context, tx *sql.Tx, metric string, w CatalogWindow) ([]HourPair, error) {
+	rows, err := tx.QueryContext(ctx, commonHoursQuery,
+		metric, w.Coarse, metric, w.Fine, FormatTime(w.Horizon), w.Hours)
+	if err != nil {
+		return nil, fmt.Errorf("select common hours: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	out := make([]HourPair, 0, w.Hours)
+	for rows.Next() {
+		var p HourPair
+		var raw string
+		if err := rows.Scan(&raw, &p.FinePoints, &p.FineUnits, &p.CoarsePoints, &p.CoarseUnits); err != nil {
+			return nil, fmt.Errorf("scan common hour: %w", err)
+		}
+		if p.Hour, err = ParseTime(raw); err != nil {
+			return nil, err
+		}
+		out = append(out, p)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("select common hours: %w", err)
+	}
+	return out, nil
+}
+
+// readHourPairs дочитывает содержимое объектов — ОДНИМ запросом и только у
+// часов, прошедших отбор по учётным колонкам.
+func readHourPairs(ctx context.Context, tx *sql.Tx, metric string, w CatalogWindow, pairs []HourPair) error {
+	at := make(map[string]int, len(pairs))
+	args := make([]any, 0, len(pairs)+3)
+	args = append(args, metric, w.Fine, w.Coarse)
+	for i := range pairs {
+		if pairs[i].CoarsePoints != w.CoarsePoints || pairs[i].FinePoints < w.MinFinePoints {
+			continue
+		}
+		key := FormatTime(pairs[i].Hour)
+		at[key] = i
+		args = append(args, key)
+	}
+	if len(at) == 0 {
+		return nil
+	}
+
+	rows, err := tx.QueryContext(ctx, hourPairsQuery(len(at)), args...)
+	if err != nil {
+		return fmt.Errorf("select hour pairs: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	for rows.Next() {
+		var layer, hour string
+		var payload []byte
+		if err := rows.Scan(&layer, &hour, &payload); err != nil {
+			return fmt.Errorf("scan hour pair: %w", err)
+		}
+		i, ok := at[hour]
+		if !ok {
+			continue
+		}
+		points, err := decodePayload(payload)
+		if err != nil {
+			return err
+		}
+		if layer == w.Fine {
+			pairs[i].Fine = points
+		} else {
+			pairs[i].Coarse = points
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return fmt.Errorf("select hour pairs: %w", err)
+	}
+	return nil
+}