From f57b43005231f1c57ec8a4e7720c2cef26c324e1 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Mon, 25 May 2026 10:58:01 +0700 Subject: [PATCH] feat(db): backfill prefix snapshots and switch read path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backfill миграция; чтение префиксов через snapshot; запись без дублирования rows. Co-authored-by: Cursor --- internal/repository/postgres.go | 58 ++---- .../repository/postgres_prefix_snapshot.go | 186 ++++++++++++++++++ internal/repository/postgres_seed.go | 15 +- .../000017_prefix_snapshot_backfill.down.sql | 3 + .../000017_prefix_snapshot_backfill.up.sql | 31 +++ .../000017_prefix_snapshot_backfill.down.sql | 3 + .../000017_prefix_snapshot_backfill.up.sql | 30 +++ 7 files changed, 286 insertions(+), 40 deletions(-) create mode 100644 internal/repository/postgres_prefix_snapshot.go create mode 100644 migrations/postgres/000017_prefix_snapshot_backfill.down.sql create mode 100644 migrations/postgres/000017_prefix_snapshot_backfill.up.sql create mode 100644 migrations/sqlite/000017_prefix_snapshot_backfill.down.sql create mode 100644 migrations/sqlite/000017_prefix_snapshot_backfill.up.sql diff --git a/internal/repository/postgres.go b/internal/repository/postgres.go index 42f0af8..1be39b5 100644 --- a/internal/repository/postgres.go +++ b/internal/repository/postgres.go @@ -800,6 +800,9 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri } return nil, "", false } + if snapID, ok := p.revisionPrefixSnapshotID(ctx, revisionID); ok { + return p.listSnapshotPrefixes(ctx, snapID, cursor, limit) + } var rows pgx.Rows var err error if useOffset { @@ -884,11 +887,7 @@ func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (st if err := copyRevisionPreview(ctx, tx, newID, sourceRevisionID); err != nil { return "", err } - _, err = tx.Exec(ctx, ` - INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source, meta_json) - SELECT $1::uuid, prefix, community_id, source, meta_json FROM revision_materialized_prefix WHERE revision_id=$2::uuid`, - newID, sourceRevisionID) - if err != nil { + if err := p.copyRevisionPrefixSnapshotRef(ctx, tx, newID, sourceRevisionID); err != nil { return "", err } if err := tx.Commit(ctx); err != nil { @@ -909,19 +908,17 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro ctx := context.Background() var unchanged int err := p.pool.QueryRow(ctx, ` - SELECT COUNT(*)::int FROM revision_materialized_prefix b - INNER JOIN revision_materialized_prefix a - ON a.revision_id = $1::uuid AND a.prefix = b.prefix - WHERE b.revision_id = $2::uuid`, aID, bID).Scan(&unchanged) + SELECT COUNT(*)::int FROM ( + SELECT b.prefix FROM (`+sqlRevisionPrefixes("$2")+`) b + INNER JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix + ) t`, aID, bID).Scan(&unchanged) if err != nil { return nil, err } rowsAdded, err := p.pool.Query(ctx, ` - SELECT b.prefix::text - FROM revision_materialized_prefix b - LEFT JOIN revision_materialized_prefix a - ON a.revision_id = $1::uuid AND a.prefix = b.prefix - WHERE b.revision_id = $2::uuid AND a.prefix IS NULL + SELECT b.prefix::text FROM (`+sqlRevisionPrefixes("$2")+`) b + LEFT JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix + WHERE a.prefix IS NULL ORDER BY b.prefix LIMIT $3`, aID, bID, maxRevisionDiffRows+1) if err != nil { @@ -942,11 +939,9 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro } addedTruncated := len(added) >= maxRevisionDiffRows rowsRem, err := p.pool.Query(ctx, ` - SELECT a.prefix::text - FROM revision_materialized_prefix a - LEFT JOIN revision_materialized_prefix b - ON b.revision_id = $2::uuid AND b.prefix = a.prefix - WHERE a.revision_id = $1::uuid AND b.prefix IS NULL + SELECT a.prefix::text FROM (`+sqlRevisionPrefixes("$1")+`) a + LEFT JOIN (`+sqlRevisionPrefixes("$2")+`) b ON b.prefix = a.prefix + WHERE b.prefix IS NULL ORDER BY a.prefix LIMIT $3`, aID, bID, maxRevisionDiffRows+1) if err != nil { @@ -1109,25 +1104,12 @@ func (p *Postgres) CreateRenderRevision(revisionID, tenantID, moduleID string, p if err := insertRevisionPreview(ctx, tx, revID, previewFragments); err != nil { return err } - if len(prefixes) > 0 { - _, err = tx.CopyFrom(ctx, - pgx.Identifier{"revision_materialized_prefix"}, - []string{"revision_id", "prefix", "community_id", "source"}, - pgx.CopyFromSlice(len(prefixes), func(i int) ([]any, error) { - pr := prefixes[i] - var comm any - if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" { - comm = strings.TrimSpace(*pr.CommunityID) - } - src := pr.Source - if strings.TrimSpace(src) == "" { - src = "render" - } - return []any{revID, strings.TrimSpace(pr.Prefix), comm, src}, nil - })) - if err != nil { - return err - } + snapID, err := p.ensurePrefixSnapshot(ctx, tx, contentHash, prefixes) + if err != nil { + return err + } + if err := p.linkRevisionPrefixSnapshot(ctx, tx, revID, snapID); err != nil { + return err } if err := tx.Commit(ctx); err != nil { return err diff --git a/internal/repository/postgres_prefix_snapshot.go b/internal/repository/postgres_prefix_snapshot.go new file mode 100644 index 0000000..c9d7166 --- /dev/null +++ b/internal/repository/postgres_prefix_snapshot.go @@ -0,0 +1,186 @@ +package repository + +import ( + "context" + "errors" + "strings" + + "evobgp/internal/store" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +func prefixSnapshotTableExists(ctx context.Context, q queryRower) bool { + var n int + err := q.QueryRow(ctx, ` + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'prefix_snapshot' + LIMIT 1`).Scan(&n) + return err == nil +} + +func normalizeSnapshotHash(contentHash string) string { + h := strings.TrimSpace(contentHash) + if strings.HasPrefix(h, "sha256:") { + h = strings.TrimPrefix(h, "sha256:") + } + if len(h) > 64 { + h = h[:64] + } + if len(h) < 64 { + h = h + strings.Repeat("0", 64-len(h)) + } + return h +} + +func (p *Postgres) revisionPrefixSnapshotID(ctx context.Context, revisionID string) (string, bool) { + if !prefixSnapshotTableExists(ctx, p.pool) { + return "", false + } + var snap *string + err := p.pool.QueryRow(ctx, ` + SELECT prefix_snapshot_id::text FROM config_revision + WHERE id = $1::uuid AND prefix_snapshot_id IS NOT NULL`, revisionID).Scan(&snap) + if err != nil || snap == nil || strings.TrimSpace(*snap) == "" { + return "", false + } + return *snap, true +} + +func (p *Postgres) ensurePrefixSnapshot(ctx context.Context, db execQuerier, contentHash string, prefixes []store.PrefixRow) (string, error) { + if !prefixSnapshotTableExists(ctx, db) { + return "", nil + } + hash := normalizeSnapshotHash(contentHash) + var existing string + err := db.QueryRow(ctx, `SELECT id::text FROM prefix_snapshot WHERE content_hash = $1`, hash).Scan(&existing) + if err == nil && existing != "" { + return existing, nil + } + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + return "", err + } + snapID := uuid.NewString() + if _, err := db.Exec(ctx, ` + INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2) + ON CONFLICT (content_hash) DO NOTHING`, snapID, hash); err != nil { + return "", err + } + if err := db.QueryRow(ctx, `SELECT id::text FROM prefix_snapshot WHERE content_hash = $1`, hash).Scan(&snapID); err != nil { + return "", err + } + var rowCount int + _ = db.QueryRow(ctx, `SELECT COUNT(*)::int FROM prefix_snapshot_row WHERE snapshot_id = $1::uuid`, snapID).Scan(&rowCount) + if rowCount > 0 { + return snapID, nil + } + for i, pr := range prefixes { + var comm any + if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" { + comm = strings.TrimSpace(*pr.CommunityID) + } + src := pr.Source + if strings.TrimSpace(src) == "" { + src = "render" + } + if _, err := db.Exec(ctx, ` + INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source) + VALUES ($1::uuid, $2, $3::cidr, $4::uuid, $5)`, + snapID, i, strings.TrimSpace(pr.Prefix), comm, src); err != nil { + return "", err + } + } + return snapID, nil +} + +func (p *Postgres) listSnapshotPrefixes(ctx context.Context, snapshotID, cursor string, limit int) ([]store.PrefixRow, string, bool) { + afterOrd, off, useOffset := store.ParsePrefixPageCursor(cursor) + var rows pgx.Rows + var err error + if useOffset { + rows, err = p.pool.Query(ctx, ` + SELECT ord, prefix::text, community_id::text, source + FROM prefix_snapshot_row + WHERE snapshot_id = $1::uuid + ORDER BY ord + LIMIT $2 OFFSET $3`, snapshotID, limit+1, off) + } else { + var afterArg any + if afterOrd != nil { + afterArg = int(*afterOrd) + } + rows, err = p.pool.Query(ctx, ` + SELECT ord, prefix::text, community_id::text, source + FROM prefix_snapshot_row + WHERE snapshot_id = $1::uuid AND ($2::int IS NULL OR ord > $2::int) + ORDER BY ord + LIMIT $3`, snapshotID, afterArg, limit+1) + } + if err != nil { + return nil, "", false + } + defer rows.Close() + var all []store.PrefixRow + var ords []int64 + for rows.Next() { + var ord int + var pr store.PrefixRow + var comm *string + if err := rows.Scan(&ord, &pr.Prefix, &comm, &pr.Source); err != nil { + continue + } + pr.CommunityID = comm + ords = append(ords, int64(ord)) + all = append(all, pr) + } + more := len(all) > limit + if more { + all = all[:limit] + ords = ords[:limit] + } + next := "" + if more && len(ords) > 0 { + next = store.FormatPrefixPageCursor(ords[len(ords)-1]) + } + if len(all) == 0 { + return nil, "", false + } + return all, next, more +} + +func (p *Postgres) linkRevisionPrefixSnapshot(ctx context.Context, db execQuerier, revisionID, snapshotID string) error { + if snapshotID == "" || !prefixSnapshotTableExists(ctx, db) { + return nil + } + _, err := db.Exec(ctx, ` + UPDATE config_revision SET prefix_snapshot_id = $2::uuid WHERE id = $1::uuid`, + revisionID, snapshotID) + return err +} + +func (p *Postgres) copyRevisionPrefixSnapshotRef(ctx context.Context, db execQuerier, dstRevisionID, srcRevisionID string) error { + if !prefixSnapshotTableExists(ctx, db) { + return nil + } + _, err := db.Exec(ctx, ` + UPDATE config_revision dst + SET prefix_snapshot_id = src.prefix_snapshot_id + FROM config_revision src + WHERE dst.id = $1::uuid AND src.id = $2::uuid AND src.prefix_snapshot_id IS NOT NULL`, + dstRevisionID, srcRevisionID) + return err +} + +func sqlRevisionPrefixes(revParam string) string { + return `SELECT psr.prefix FROM config_revision cr +JOIN prefix_snapshot_row psr ON psr.snapshot_id = cr.prefix_snapshot_id +WHERE cr.id = ` + revParam + `::uuid AND cr.prefix_snapshot_id IS NOT NULL +UNION ALL +SELECT rmp.prefix FROM revision_materialized_prefix rmp +WHERE rmp.revision_id = ` + revParam + `::uuid + AND NOT EXISTS ( + SELECT 1 FROM config_revision cr2 + WHERE cr2.id = ` + revParam + `::uuid AND cr2.prefix_snapshot_id IS NOT NULL + )` +} diff --git a/internal/repository/postgres_seed.go b/internal/repository/postgres_seed.go index 431b4e6..e3dc45c 100644 --- a/internal/repository/postgres_seed.go +++ b/internal/repository/postgres_seed.go @@ -85,9 +85,20 @@ protocol direct { INSERT INTO config_revision_preview (revision_id, fragments) VALUES ($1::uuid, $2::jsonb)`, rid, string(previewFragsB)); err != nil { return err } + snapID := uuid.NewString() + demoHash := normalizeSnapshotHash("sha256:demo-rev-1") if _, err := tx.Exec(ctx, ` - INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source) - VALUES ($1::uuid,'203.0.113.0/24',$2::uuid,'demo'), ($1::uuid,'2001:db8::/32',$2::uuid,'demo')`, rid, cid); err != nil { + INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2)`, snapID, demoHash); err != nil { + return err + } + if _, err := tx.Exec(ctx, ` + UPDATE config_revision SET prefix_snapshot_id = $2::uuid WHERE id = $1::uuid`, rid, snapID); err != nil { + return err + } + if _, err := tx.Exec(ctx, ` + INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source) + VALUES ($1::uuid, 0, '203.0.113.0/24'::cidr, $2::uuid, 'demo'), + ($1::uuid, 1, '2001:db8::/32'::cidr, $2::uuid, 'demo')`, snapID, cid); err != nil { return err } if _, err := tx.Exec(ctx, ` diff --git a/migrations/postgres/000017_prefix_snapshot_backfill.down.sql b/migrations/postgres/000017_prefix_snapshot_backfill.down.sql new file mode 100644 index 0000000..0fc540a --- /dev/null +++ b/migrations/postgres/000017_prefix_snapshot_backfill.down.sql @@ -0,0 +1,3 @@ +DELETE FROM prefix_snapshot_row; +UPDATE config_revision SET prefix_snapshot_id = NULL WHERE prefix_snapshot_id IS NOT NULL; +DELETE FROM prefix_snapshot; diff --git a/migrations/postgres/000017_prefix_snapshot_backfill.up.sql b/migrations/postgres/000017_prefix_snapshot_backfill.up.sql new file mode 100644 index 0000000..ba1f3c4 --- /dev/null +++ b/migrations/postgres/000017_prefix_snapshot_backfill.up.sql @@ -0,0 +1,31 @@ +-- Backfill prefix snapshots from revision_materialized_prefix. + +WITH new_snaps AS ( + INSERT INTO prefix_snapshot (id, content_hash) + SELECT gen_random_uuid(), + substr(replace(cr.id::text, '-', '') || replace(cr.id::text, '-', ''), 1, 64) + FROM config_revision cr + WHERE cr.prefix_snapshot_id IS NULL + AND EXISTS ( + SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = cr.id + ) + RETURNING id, content_hash +) +UPDATE config_revision cr +SET prefix_snapshot_id = ns.id +FROM new_snaps ns +WHERE cr.prefix_snapshot_id IS NULL + AND ns.content_hash = substr(replace(cr.id::text, '-', '') || replace(cr.id::text, '-', ''), 1, 64); + +INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source) +SELECT cr.prefix_snapshot_id, + (row_number() OVER (PARTITION BY cr.id ORDER BY rmp.id) - 1)::int, + rmp.prefix, + rmp.community_id, + rmp.source +FROM config_revision cr +JOIN revision_materialized_prefix rmp ON rmp.revision_id = cr.id +WHERE cr.prefix_snapshot_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM prefix_snapshot_row psr WHERE psr.snapshot_id = cr.prefix_snapshot_id + ); diff --git a/migrations/sqlite/000017_prefix_snapshot_backfill.down.sql b/migrations/sqlite/000017_prefix_snapshot_backfill.down.sql new file mode 100644 index 0000000..de9e583 --- /dev/null +++ b/migrations/sqlite/000017_prefix_snapshot_backfill.down.sql @@ -0,0 +1,3 @@ +DELETE FROM prefix_snapshot_row; +UPDATE config_revision SET prefix_snapshot_id = NULL; +DELETE FROM prefix_snapshot; diff --git a/migrations/sqlite/000017_prefix_snapshot_backfill.up.sql b/migrations/sqlite/000017_prefix_snapshot_backfill.up.sql new file mode 100644 index 0000000..d130223 --- /dev/null +++ b/migrations/sqlite/000017_prefix_snapshot_backfill.up.sql @@ -0,0 +1,30 @@ +-- SQLite backfill: one snapshot per revision with materialized prefixes. + +INSERT INTO prefix_snapshot (id, content_hash) +SELECT lower(hex(randomblob(16))), + substr(replace(cr.id, '-', '') || replace(cr.id, '-', ''), 1, 64) +FROM config_revision cr +WHERE cr.prefix_snapshot_id IS NULL + AND EXISTS (SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = cr.id); + +UPDATE config_revision +SET prefix_snapshot_id = ( + SELECT ps.id FROM prefix_snapshot ps + WHERE ps.content_hash = substr(replace(config_revision.id, '-', '') || replace(config_revision.id, '-', ''), 1, 64) +) +WHERE prefix_snapshot_id IS NULL + AND EXISTS (SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = config_revision.id); + +INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source) +SELECT cr.prefix_snapshot_id, + (SELECT COUNT(*) FROM revision_materialized_prefix r2 + WHERE r2.revision_id = cr.id AND r2.id <= rmp.id) - 1, + rmp.prefix, + rmp.community_id, + rmp.source +FROM config_revision cr +JOIN revision_materialized_prefix rmp ON rmp.revision_id = cr.id +WHERE cr.prefix_snapshot_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 FROM prefix_snapshot_row psr WHERE psr.snapshot_id = cr.prefix_snapshot_id + );