Senior

/* Es gibt zwei PostgreSQL-Server: * PROD - OLTP-Server, * STATS - Server für lange analytische Abfragen. Auf dem aktuellen Server in der Datenbank prod gibt es eine große Tabelle (10Tb) mit folgendem Aufbau: CREATE TABLE profiles( id SERIAL, data JSONB ) In der Tabelle können "Lücken" sein, d.h. einige `id` können fehlen. Es ist notwendig, ein Programm zu schreiben, um die Tabelle profiles von PROD nach STATS zu kopieren. Es wird angenommen, dass die folgenden Schnittstellen für die Arbeit mit Datenbanken verwendet werden: type Row []interface{} type Database interface { // die Implementierung des Database-Interfaces kann Verbindungen neu herstellen // der Aufruf von SaveRows ist idempotent io.Closer GetMaxID(ctx context.Context) (uint64, error) LoadRows(ctx context.Context, minID, maxID uint64) ([]Row, error) // [minID, maxID] SaveRows(ctx context.Context, rows []Row) error } func Connect(ctx context.Context, dbname string) (Database, error) // CopyTable // Wenn full=false, Datenübertragung ab der Stelle des vorherigen Fehlers fortsetzen // Wenn full=true, alle Daten übertragen func CopyTable(fromName string, toName string, full bool) error { // ... dein Code } Wenn die Option `full=false` übergeben wird, soll das Programm die Datenübertragung ab der Stelle des vorherigen Fehlers fortsetzen. Wenn `full=true`, sollen alle Daten übertragen werden. Basisebene: - sequenzielle Datenübertragung in einem einzigen Thread - Fehlerbehebung (Option `full=false`) Zusätzliche Informationen: - falls notwendig, kann die Schnittstelle erweitert werden, indem eigene Methoden hinzugefügt werden - falls notwendig, kann direkt das Paket database/sql verwendet werden

sobes.tech KI

Antwort von AI

Для решения задачи копирования большой таблицы profiles с сервера PROD на STATS с возможностью продолжения после сбоев и опцией полного копирования, можно реализовать следующий подход:

Определить последний успешно скопированный ID на сервере STATS. Для этого можно добавить в интерфейс Database метод, например, GetMaxCopiedID(ctx context.Context) (uint64, error), который вернёт максимальный id, уже скопированный в таблицу на сервере STATS.
При запуске функции CopyTable:
- Если full=true, то начинаем копирование с id=1 (или минимального id в таблице PROD).
- Если full=false, то запрашиваем последний скопированный id на STATS и начинаем копирование с id = lastCopiedID + 1.
Для копирования данных используем батчи по диапазонам id, например, по 1000 записей за раз. В цикле:
- Получаем максимальный id в таблице PROD через GetMaxID.
- Загружаем строки из PROD с id в диапазоне [startID, startID+batchSize-1] через LoadRows.
- Сохраняем их на STATS через SaveRows.
- Обновляем startID = startID + batchSize.
Если во время копирования происходит сбой, при следующем запуске с full=false процесс возобновится с последнего успешно сохранённого id.

Пример расширения интерфейса и реализации CopyTable:

// Расширение интерфейса Database для получения максимального скопированного id
func (db Database) GetMaxCopiedID(ctx context.Context) (uint64, error) {
    // Реализация зависит от структуры таблицы на STATS
    // Например, SELECT MAX(id) FROM profiles
}

func CopyTable(fromName string, toName string, full bool) error {
    ctx := context.Background()
    fromDB, err := Connect(ctx, fromName)
    if err != nil {
        return err
    }
    defer fromDB.Close()

    toDB, err := Connect(ctx, toName)
    if err != nil {
        return err
    }
    defer toDB.Close()

    var startID uint64 = 1
    if !full {
        startID, err = toDB.GetMaxCopiedID(ctx)
        if err != nil {
            return err
        }
        startID++
    }

    maxID, err := fromDB.GetMaxID(ctx)
    if err != nil {
        return err
    }

    batchSize := uint64(1000)

    for currentID := startID; currentID <= maxID; currentID += batchSize {
        endID := currentID + batchSize - 1
        if endID > maxID {
            endID = maxID
        }

        rows, err := fromDB.LoadRows(ctx, currentID, endID)
        if err != nil {
            return err
        }

        if len(rows) == 0 {
            continue
        }

        err = toDB.SaveRows(ctx, rows)
        if err != nil {
            return err
        }
    }

    return nil
}

Таким образом, программа последовательно переливает данные, восстанавливается после сбоев и поддерживает опцию полного копирования.

Zurück Als nächstes