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 }