SH

Sascha Hess

xenosystems.de - IT-Consulting & Data Management

www.xenosystems.de

STRATEGISCHE WISSENS-ROADMAP 2026

SQL Server

Datenmigration & Systemwechsel

Migrationen planen, durchführen und validieren — ohne Datenverlust und ohne Betriebsunterbrechung

1

Migrations-Phasenmodell

Assessment, Vorbereitung, Migration, Validierung, Cutover — strukturiert

2

10 Migrations-Fallen

Die häufigsten Fehler bei SQL-Datenmigrationsprojekten — mit Korrekturen

3

Fertige Migrations-Skripte

Schema-Vergleich, Datentransfer, Validierung und Rollback — sofort einsetzbar

4

Validierungs-Framework

Row Counts, Checksums, Business-Reconciliation und Integritätsprüfungen

5

30-Tage-Migrationsplan

Von der Analyse bis zum erfolgreichen Cutover — strukturiert und risikoarm

DIE VIER HÄUFIGSTEN MIGRATIONS-AUSLÖSER IM KMU

■  ERP-Ablösung: Das Altsystem wird abgekündigt oder nicht mehr supported — Zeitdruck ist vorprogrammiert.

■  SQL Server End of Support: SQL Server 2012 und 2014 sind seit Jahren out of support — viele KMU nutzen sie noch.

■  Hardware-Ablösung: Server-Lifecycle-Ende zwingt zur Migration — oft mit zu wenig Vorlaufzeit geplant.

■  Cloud-Strategie: Azure SQL oder Managed Instance als Ziel einer strategischen Cloud-Entscheidung.

1. Assessment

Vollständiges Bild des Ist-Zustands

20 %

Inventar, Risikobewertung

2. Vorbereitung

Zielstruktur, Mapping, Testumgebung

25 %

DDL-Skripte, Mapping-Tabellen

3. Migration (Test)

Mehrere Testmigrationen, Fehler beheben

30 %

Migrations-Skripte, Protokoll

4. Validierung

Vollständige Korrektheitsprüfung

15 %

Validierungsberichte

5. Cutover

Produktiv-Migration, Go-Live, Monitoring

10 %

Abnahmeprotokoll

-- Vollständiges Datenbank-Inventar für die Quell-Datenbank

SELECT

    SCHEMA_NAME(o.schema_id)    AS Schema_Name,

    o.name                      AS Objekt_Name,

    o.type_desc                 AS Objekt_Typ,

    o.create_date               AS Erstellt,

    o.modify_date               AS Geaendert,

    -- Zeilen-Anzahl für Tabellen

    CASE WHEN o.type = 'U'

         THEN (SELECT SUM(p.rows) FROM sys.partitions p

               WHERE p.object_id = o.object_id AND p.index_id IN (0,1))

         ELSE NULL END          AS Zeilen_Anzahl,

    -- Größe in MB

    CASE WHEN o.type = 'U'

         THEN ROUND(

            (SELECT SUM(a.total_pages) * 8.0 / 1024

             FROM sys.indexes i

             JOIN sys.partitions p ON i.object_id = p.object_id

             JOIN sys.allocation_units a ON p.partition_id = a.container_id

             WHERE i.object_id = o.object_id), 2)

         ELSE NULL END          AS Groesse_MB,

    -- Abhängigkeiten vorhanden?

    (SELECT COUNT(*) FROM sys.sql_expression_dependencies d

     WHERE d.referenced_id = o.object_id)  AS Wird_Referenziert_Von,

    (SELECT COUNT(*) FROM sys.sql_expression_dependencies d

     WHERE d.referencing_id = o.object_id) AS Referenziert_Objekte

FROM sys.objects o

WHERE o.is_ms_shipped = 0

ORDER BY o.type_desc, Schema_Name, o.name;

-- Aktuellen Kompatibilitäts-Level prüfen

SELECT name, compatibility_level,

    CASE compatibility_level

        WHEN 170 THEN 'SQL Server 2025 — aktuell'

        WHEN 160 THEN 'SQL Server 2022 — gut'

        WHEN 150 THEN 'SQL Server 2019 — veraltet'

        WHEN 140 THEN 'SQL Server 2017 — veraltet'

        WHEN 130 THEN 'SQL Server 2016 — kritisch veraltet'

        ELSE 'Sehr alt — sofort aktualisieren'

    END AS Bewertung

FROM sys.databases WHERE database_id > 4;

-- Schrittweise Anhebung mit Query Store-Überwachung

ALTER DATABASE [MeineDatenbank] SET QUERY_STORE = ON

(OPERATION_MODE = READ_WRITE, QUERY_CAPTURE_MODE = AUTO, MAX_STORAGE_SIZE_MB = 500);

-- Kompatibilitäts-Level anheben

ALTER DATABASE [MeineDatenbank] SET COMPATIBILITY_LEVEL = 170;

-- Regressions-Analyse nach der Anhebung (nächster Tag)

SELECT TOP 20

    qsq.query_id,

    qsrs.avg_duration / 1000.0          AS Avg_Dauer_ms_Neu,

    qsrs_alt.avg_duration / 1000.0      AS Avg_Dauer_ms_Alt,

    (qsrs.avg_duration - qsrs_alt.avg_duration) /

        qsrs_alt.avg_duration * 100.0   AS Verschlechterung_Pct,

    SUBSTRING(qsqt.query_sql_text,1,200) AS Query_Text

FROM sys.query_store_query qsq

JOIN sys.query_store_query_text qsqt ON qsq.query_text_id = qsqt.query_text_id

JOIN sys.query_store_plan qsp ON qsq.query_id = qsp.query_id

JOIN sys.query_store_runtime_stats qsrs ON qsp.plan_id = qsrs.plan_id

JOIN sys.query_store_runtime_stats_interval qsrsi

    ON qsrs.runtime_stats_interval_id = qsrsi.runtime_stats_interval_id

-- Vergleich mit Stats vor Kompatibilitäts-Level-Änderung

JOIN sys.query_store_runtime_stats qsrs_alt ON qsq.query_id =

    (SELECT TOP 1 q2.query_id FROM sys.query_store_query q2

     WHERE q2.query_text_id = qsq.query_text_id)

WHERE qsrsi.start_time > DATEADD(DAY,-1,GETDATE())

  AND qsrs.avg_duration > qsrs_alt.avg_duration * 1.3  -- 30% Verschlechterung

ORDER BY Verschlechterung_Pct DESC;

-- Alle deaktivierten Constraints finden — kritischer Befund

SELECT

    SCHEMA_NAME(t.schema_id)        AS Schema_Name,

    t.name                          AS Tabelle,

    c.name                          AS Constraint_Name,

    c.type_desc                     AS Constraint_Typ,

    c.is_disabled                   AS Deaktiviert,

    c.is_not_trusted                AS Nicht_Vertrauenswuerdig,

    -- is_not_trusted = 1 bedeutet: Constraint existiert, aber Daten wurden nie geprüft

    CASE WHEN c.is_disabled = 1 THEN 'KRITISCH: Deaktiviert'

         WHEN c.is_not_trusted = 1 THEN 'WARNUNG: Nicht verifiziert'

         ELSE 'OK'

    END AS Status

FROM sys.check_constraints c

JOIN sys.tables t ON c.parent_object_id = t.object_id

UNION ALL

SELECT

    SCHEMA_NAME(t.schema_id), t.name, f.name, 'FOREIGN_KEY',

    f.is_disabled, f.is_not_trusted,

    CASE WHEN f.is_disabled = 1 THEN 'KRITISCH: Deaktiviert'

         WHEN f.is_not_trusted = 1 THEN 'WARNUNG: Nicht verifiziert'

         ELSE 'OK' END

FROM sys.foreign_keys f

JOIN sys.tables t ON f.parent_object_id = t.object_id

WHERE f.is_disabled = 1 OR f.is_not_trusted = 1

ORDER BY Status, Schema_Name, Tabelle;

-- Constraints korrekt wieder aktivieren (mit vollständiger Prüfung)

-- Schlägt fehl wenn inkonsistente Daten vorhanden — das ist gewollt!

ALTER TABLE dbo.Auftragspositionen

    WITH CHECK CHECK CONSTRAINT FK_Auftraege_Positionen;

-- Datentypvergleich Quelle vs. Ziel

SELECT

    src.TABLE_SCHEMA                AS Schema,

    src.TABLE_NAME                  AS Tabelle,

    src.COLUMN_NAME                 AS Spalte,

    src.DATA_TYPE                   AS Quell_Typ,

    src.CHARACTER_MAXIMUM_LENGTH    AS Quell_Laenge,

    tgt.DATA_TYPE                   AS Ziel_Typ,

    tgt.CHARACTER_MAXIMUM_LENGTH    AS Ziel_Laenge,

    -- Risikobewertung

    CASE

        WHEN src.DATA_TYPE <> tgt.DATA_TYPE THEN 'KRITISCH: Typ unterschiedlich'

        WHEN src.CHARACTER_MAXIMUM_LENGTH > tgt.CHARACTER_MAXIMUM_LENGTH

        THEN 'HOCH: Ziel-Spalte kürzer — Abschneidung möglich'

        WHEN tgt.CHARACTER_MAXIMUM_LENGTH > src.CHARACTER_MAXIMUM_LENGTH

        THEN 'INFO: Ziel-Spalte größer — OK'

        ELSE 'OK'

    END AS Risiko

FROM INFORMATION_SCHEMA.COLUMNS src

JOIN [Ziel-Server].[Ziel-DB].INFORMATION_SCHEMA.COLUMNS tgt

    ON src.TABLE_SCHEMA = tgt.TABLE_SCHEMA

    AND src.TABLE_NAME  = tgt.TABLE_NAME

    AND src.COLUMN_NAME = tgt.COLUMN_NAME

WHERE src.TABLE_CATALOG = 'Quell-DB'

ORDER BY Risiko DESC, Schema, Tabelle, Spalte;

-- Alle SQL-Logins mit Passwort-Hash scripen (für Migration)

SELECT

    'CREATE LOGIN ' + QUOTENAME(name) +

    ' WITH PASSWORD = ' + CONVERT(NVARCHAR(MAX),

        LOGINPROPERTY(name, 'PasswordHash'), 1) +

    ' HASHED, SID = ' + CONVERT(NVARCHAR(MAX), sid, 1) +

    ', CHECK_POLICY = ' +

    CASE WHEN is_policy_checked = 1 THEN 'ON' ELSE 'OFF' END +

    ';'                             AS Login_Script

FROM sys.sql_logins

WHERE name NOT LIKE '##%'

  AND name NOT IN ('sa')

ORDER BY name;

-- Verwaiste Datenbankbenutzer finden und reparieren

SELECT dp.name AS Benutzer, dp.sid, dp.type_desc

FROM sys.database_principals dp

WHERE dp.type IN ('S','U')

  AND dp.authentication_type_desc = 'INSTANCE'

  AND NOT EXISTS (SELECT 1 FROM sys.server_principals sp WHERE sp.sid = dp.sid);

-- Reparieren: ALTER USER [Username] WITH LOGIN = [LoginName];

-- SQL Agent Jobs scripen (für manuelle Übertragung oder SSIS-Transfer)

SELECT

    j.name                          AS Job_Name,

    j.enabled                       AS Aktiv,

    j.description,

    c.name                          AS Kategorie,

    s.name                          AS Besitzer,

    -- Zeitplan

    sc.name                         AS Zeitplan,

    sc.freq_type, sc.freq_interval,

    -- Letzter Lauf

    jh.run_date, jh.run_time, jh.run_status

FROM msdb.dbo.sysjobs j

LEFT JOIN msdb.dbo.syscategories c ON j.category_id = c.category_id

LEFT JOIN sys.server_principals s ON j.owner_sid = s.sid

LEFT JOIN msdb.dbo.sysjobschedules js ON j.job_id = js.job_id

LEFT JOIN msdb.dbo.sysschedules sc ON js.schedule_id = sc.schedule_id

LEFT JOIN msdb.dbo.sysjobhistory jh ON j.job_id = jh.job_id

    AND jh.step_id = 0

    AND jh.run_date = (SELECT MAX(run_date) FROM msdb.dbo.sysjobhistory

                       WHERE job_id = j.job_id AND step_id = 0)

ORDER BY j.name;

-- Vollständiger Datenbank-Fingerprint in unter 5 Minuten

SELECT

    -- Größe und Volumen

    DB_NAME()                           AS Datenbank,

    (SELECT COUNT(*) FROM sys.tables)   AS Tabellen_Anzahl,

    (SELECT COUNT(*) FROM sys.views)    AS Views_Anzahl,

    (SELECT COUNT(*) FROM sys.procedures) AS Procs_Anzahl,

    (SELECT COUNT(*) FROM sys.triggers) AS Trigger_Anzahl,

    (SELECT COUNT(*) FROM sys.indexes WHERE is_primary_key = 0

        AND type > 0)                   AS Index_Anzahl,

    (SELECT COUNT(*) FROM sys.foreign_keys) AS FK_Anzahl,

    (SELECT COUNT(*) FROM sys.check_constraints) AS Check_Constraints,

    -- Gesamtgröße

    (SELECT ROUND(SUM(size * 8.0 / 1024), 0) FROM sys.database_files

     WHERE type_desc = 'ROWS')          AS Datei_MB,

    (SELECT ROUND(SUM(size * 8.0 / 1024), 0) FROM sys.database_files

     WHERE type_desc = 'LOG')           AS Log_MB,

    -- Tatsächlich belegter Platz

    (SELECT ROUND(SUM(a.total_pages) * 8.0 / 1024, 0)

     FROM sys.allocation_units a)       AS Belegt_MB,

    -- Kompatibilitätslevel

    (SELECT compatibility_level FROM sys.databases WHERE database_id = DB_ID()) AS Compat_Level,

    -- Features die migriert werden müssen

    (SELECT COUNT(*) FROM sys.linked_servers) AS Linked_Server,

    (SELECT COUNT(*) FROM sys.certificates) AS Certificates,

    (SELECT COUNT(*) FROM msdb.dbo.sysjobs j

     JOIN msdb.dbo.sysjobsteps js ON j.job_id = js.job_id

     WHERE js.database_name = DB_NAME()) AS Agent_Jobs;

-- Jede Tabelle bekommt einen Migrations-Komplexitäts-Score

-- Hilft bei der Priorisierung und Zeitplanung

SELECT

    SCHEMA_NAME(t.schema_id)            AS Schema_Name,

    t.name                              AS Tabelle,

    -- Volumen

    p.Zeilen,

    ROUND(s.Groesse_MB, 1)              AS Groesse_MB,

    -- Strukturelle Komplexität

    col.Spalten_Anzahl,

    fk_out.FK_Ausgehend,

    fk_in.FK_Eingehend,

    idx.Index_Anzahl,

    trg.Trigger_Anzahl,

    -- Datenkomplexität

    lob.LOB_Spalten,                    -- TEXT, NTEXT, IMAGE, VARBINARY(MAX)

    comp.Computed_Columns,

    -- Komplexitäts-Score (0–100)

    LEAST(100,

        CASE WHEN p.Zeilen > 10000000 THEN 30

             WHEN p.Zeilen > 1000000  THEN 20

             WHEN p.Zeilen > 100000   THEN 10

             ELSE 5 END

      + fk_in.FK_Eingehend * 3         -- Referenziert von anderen: komplex

      + fk_out.FK_Ausgehend * 2

      + ISNULL(trg.Trigger_Anzahl,0) * 5

      + ISNULL(lob.LOB_Spalten,0) * 8  -- LOB-Daten: besonders aufwendig

      + ISNULL(comp.Computed_Columns,0) * 4

    )                                   AS Migrations_Score

FROM sys.tables t

JOIN (SELECT object_id, SUM(rows) AS Zeilen

      FROM sys.partitions WHERE index_id IN (0,1)

      GROUP BY object_id) p ON t.object_id = p.object_id

JOIN (SELECT object_id, SUM(total_pages) * 8.0/1024 AS Groesse_MB

      FROM sys.indexes i JOIN sys.partitions pt ON i.object_id = pt.object_id

      JOIN sys.allocation_units au ON pt.partition_id = au.container_id

      GROUP BY object_id) s ON t.object_id = s.object_id

JOIN (SELECT object_id, COUNT(*) AS Spalten_Anzahl

      FROM sys.columns GROUP BY object_id) col ON t.object_id = col.object_id

LEFT JOIN (SELECT parent_object_id, COUNT(*) AS FK_Ausgehend

           FROM sys.foreign_keys GROUP BY parent_object_id) fk_out

    ON t.object_id = fk_out.parent_object_id

LEFT JOIN (SELECT referenced_object_id, COUNT(*) AS FK_Eingehend

           FROM sys.foreign_keys GROUP BY referenced_object_id) fk_in

    ON t.object_id = fk_in.referenced_object_id

LEFT JOIN (SELECT object_id, COUNT(*) AS Index_Anzahl

           FROM sys.indexes WHERE type > 0 AND is_primary_key = 0

           GROUP BY object_id) idx ON t.object_id = idx.object_id

LEFT JOIN (SELECT parent_id, COUNT(*) AS Trigger_Anzahl

           FROM sys.triggers GROUP BY parent_id) trg ON t.object_id = trg.parent_id

LEFT JOIN (SELECT object_id, COUNT(*) AS LOB_Spalten

           FROM sys.columns WHERE system_type_id IN (34,35,99,165,167,231,239)

           GROUP BY object_id) lob ON t.object_id = lob.object_id

LEFT JOIN (SELECT object_id, COUNT(*) AS Computed_Columns

           FROM sys.computed_columns GROUP BY object_id) comp ON t.object_id = comp.object_id

ORDER BY Migrations_Score DESC, Groesse_MB DESC;

-- Features die in der Zielversion nicht mehr unterstützt werden

-- Besonders relevant bei SQL Server 2012/2014 → 2025

-- Veraltete Datentypen

SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE,

    'VERALTET: ' + DATA_TYPE + ' → NVARCHAR(MAX) oder VARBINARY(MAX)' AS Massnahme

FROM INFORMATION_SCHEMA.COLUMNS

WHERE DATA_TYPE IN ('text','ntext','image')   -- In SQL Server 2025 noch da, aber deprecated

ORDER BY TABLE_SCHEMA, TABLE_NAME;

-- Veraltete Syntax in Stored Procedures

SELECT o.name AS Prozedur, m.definition

FROM sys.sql_modules m

JOIN sys.objects o ON m.object_id = o.object_id

WHERE m.definition LIKE '%SELECT%INTO%#%'      -- temp tables: OK

   OR m.definition LIKE '%OUTER APPLY%'        -- prüfen

   OR m.definition LIKE '%FOR XML%'            -- XML-Features prüfen

   OR m.definition LIKE '%OPENROWSET%'         -- externe Datenquellen

   OR m.definition LIKE '%sp_OACreate%'        -- OLE Automation: deprecated

   OR m.definition LIKE '%xp_cmdshell%'        -- Sicherheitsrisiko

ORDER BY o.name;

-- Linked Server-Abhängigkeiten — müssen im Ziel vorhanden sein

SELECT DISTINCT

    d.referenced_server_name   AS Linked_Server,

    COUNT(*) OVER (PARTITION BY d.referenced_server_name) AS Referenzen

FROM sys.sql_expression_dependencies d

WHERE d.referenced_server_name IS NOT NULL

ORDER BY d.referenced_server_name;

-- Schema-Migrationsskript generieren (Reihenfolge: erst Tabellen ohne FKs)

-- Für jede Tabelle: CREATE TABLE ohne FOREIGN KEY Constraints

SELECT

    'CREATE TABLE ' + QUOTENAME(SCHEMA_NAME(t.schema_id)) + '.' +

    QUOTENAME(t.name) + ' (' +

    STRING_AGG(

        QUOTENAME(c.name) + ' ' +

        tp.name +

        CASE WHEN tp.name IN ('nvarchar','varchar','char','nchar')

             THEN '(' + CASE WHEN c.max_length = -1 THEN 'MAX'

                             WHEN tp.name IN ('nvarchar','nchar')

                             THEN CAST(c.max_length/2 AS VARCHAR)

                             ELSE CAST(c.max_length AS VARCHAR) END + ')'

             WHEN tp.name IN ('decimal','numeric')

             THEN '(' + CAST(c.precision AS VARCHAR) + ',' +

                  CAST(c.scale AS VARCHAR) + ')'

             ELSE '' END +

        CASE WHEN c.is_nullable = 0 THEN ' NOT NULL' ELSE ' NULL' END +

        CASE WHEN c.is_identity = 1

             THEN ' IDENTITY(' + CAST(IDENT_SEED(t.name) AS VARCHAR) +

                  ',' + CAST(IDENT_INCR(t.name) AS VARCHAR) + ')'

             ELSE '' END,

        ', '

    ) WITHIN GROUP (ORDER BY c.column_id) + ');'

        AS DDL_Script

FROM sys.tables t

JOIN sys.columns c ON t.object_id = c.object_id

JOIN sys.types tp ON c.user_type_id = tp.user_type_id

WHERE t.is_ms_shipped = 0

GROUP BY t.schema_id, t.object_id, t.name

ORDER BY SCHEMA_NAME(t.schema_id), t.name;

-- Foreign Keys separiert — nach Datenimport hinzufügen

SELECT

    'ALTER TABLE ' + QUOTENAME(SCHEMA_NAME(t.schema_id)) + '.' +

    QUOTENAME(t.name) +

    ' ADD CONSTRAINT ' + QUOTENAME(f.name) +

    ' FOREIGN KEY (' + c_from.name + ') REFERENCES ' +

    QUOTENAME(SCHEMA_NAME(rt.schema_id)) + '.' + QUOTENAME(rt.name) +

    '(' + c_to.name + ');'  AS FK_Script

FROM sys.foreign_keys f

JOIN sys.tables t  ON f.parent_object_id   = t.object_id

JOIN sys.tables rt ON f.referenced_object_id = rt.object_id

JOIN sys.foreign_key_columns fkc ON f.object_id = fkc.constraint_object_id

JOIN sys.columns c_from ON fkc.parent_object_id   = c_from.object_id

    AND fkc.parent_column_id   = c_from.column_id

JOIN sys.columns c_to   ON fkc.referenced_object_id = c_to.object_id

    AND fkc.referenced_column_id = c_to.column_id

ORDER BY SCHEMA_NAME(t.schema_id), t.name;

Linked Server INSERT

< 10 GB

Mittel

Möglich

Einfache Migrationen innerhalb LAN

bcp (Bulk Copy)

Beliebig

Sehr hoch

Nein

Große Tabellen, maximale Performance

BULK INSERT

Beliebig

Sehr hoch

Nein

Wenn Dateien als Zwischenstufe akzeptiert

BACPAC (Azure)

< 150 GB

Mittel

Nein

SQL → Azure SQL Migration

Database Copy

Identisches Schema

Sehr hoch

Nein

SQL Server → SQL Server, gleiche Version

Log Shipping

Live-Sync

Kontinuierlich

Nein

Minimale Ausfallzeit bei Cutover

Always On AG

Live-Sync

Kontinuierlich

Nein

Enterprise: Zero-Downtime-Migration

-- Vollständige Datenmigration via Linked Server

-- Optimal für Datenbanken < 50 GB auf schnellem LAN

-- Schritt 1: Foreign Keys und Triggers deaktivieren

ALTER TABLE dbo.Auftragspositionen NOCHECK CONSTRAINT ALL;

DISABLE TRIGGER ALL ON dbo.Auftraege;

-- Schritt 2: IDENTITY_INSERT für Primärschlüssel-Übernahme

SET IDENTITY_INSERT dbo.Auftraege ON;

BEGIN TRY

    BEGIN TRANSACTION;

    -- Tabelle leeren (wenn Wiederholung möglich sein soll — Idempotenz)

    TRUNCATE TABLE dbo.Auftraege;

    -- Daten übertragen

    INSERT INTO dbo.Auftraege WITH (TABLOCK)

    (AuftragsNr, KundenNr, Auftragsdatum, Status, Nettobetrag)

    SELECT AuftragsNr, KundenNr, Auftragsdatum, Status, Nettobetrag

    FROM [Quell-Server].[Quell-DB].dbo.Auftraege;

    DECLARE @MigriertZeilen INT = @@ROWCOUNT;

    -- Sofort-Validierung: Row Count

    DECLARE @QuelleZeilen INT;

    SELECT @QuelleZeilen = COUNT(*) FROM [Quell-Server].[Quell-DB].dbo.Auftraege;

    IF @MigriertZeilen <> @QuelleZeilen

    BEGIN

        RAISERROR('Row Count Mismatch: Quelle=%d, Ziel=%d',

            16, 1, @QuelleZeilen, @MigriertZeilen);

    END

    COMMIT TRANSACTION;

    PRINT 'Auftraege migriert: ' + CAST(@MigriertZeilen AS VARCHAR) + ' Zeilen';

END TRY

BEGIN CATCH

    IF XACT_STATE() <> 0 ROLLBACK TRANSACTION;

    PRINT 'FEHLER: ' + ERROR_MESSAGE();

    THROW;

END CATCH

SET IDENTITY_INSERT dbo.Auftraege OFF;

-- Schritt 3 (nach allen Tabellen): Constraints und Triggers reaktivieren

ALTER TABLE dbo.Auftragspositionen WITH CHECK CHECK CONSTRAINT ALL;

ENABLE TRIGGER ALL ON dbo.Auftraege;

-- bcp Datenexport (in Command Shell oder PowerShell)

-- Schritt 1: Aus Quelle exportieren

-- bcp [Quell-DB].[dbo].[Auftraege] out "C:\Migration\Auftraege.bcp"

--     -S QUELL-SERVER -T -n -q

-- -T = Windows Auth, -n = native Format, -q = quoted identifiers

-- Schritt 2: In Ziel importieren

-- bcp [Ziel-DB].[dbo].[Auftraege] in "C:\Migration\Auftraege.bcp"

--     -S ZIEL-SERVER -T -n -q -b 10000 -h "TABLOCK,ORDER(AuftragsNr ASC)"

-- -b 10000 = Batch-Größe, -h TABLOCK = Tabellensperrung für Speed

-- Alternativ: BULK INSERT aus SQL Server

BULK INSERT dbo.Auftraege

FROM 'C:\Migration\Auftraege.bcp'

WITH (

    DATAFILETYPE    = 'native',

    BATCHSIZE       = 50000,        -- Commit-Intervall

    TABLOCK,                        -- Tabellen-Lock für maximale Performance

    ORDER (AuftragsNr ASC),         -- wenn Daten sortiert sind: schneller

    ROWS_PER_BATCH  = 10000000     -- Hint für Fortschrittsschätzung

);

-- Performance-Tipp: Indizes und Constraints erst nach BULK INSERT anlegen

-- Tabelle ohne Indizes befüllen ist 3-10x schneller

-- Batch-Migration: große Tabellen in Batches übertragen

-- Verhindert Log-Explosion und ermöglicht Fortsetzung bei Fehler

CREATE TABLE migration.Fortschritt (

    Tabelle         NVARCHAR(200) PRIMARY KEY,

    LetzterKey      BIGINT DEFAULT 0,

    MigriertZeilen  BIGINT DEFAULT 0,

    Status          NVARCHAR(20) DEFAULT 'NEU'

);

DECLARE @BatchSize INT = 50000;

DECLARE @LetzterKey BIGINT;

DECLARE @Importiert INT = 1;

SELECT @LetzterKey = ISNULL(LetzterKey, 0)

FROM migration.Fortschritt WHERE Tabelle = 'Auftraege';

WHILE @Importiert > 0

BEGIN

    INSERT INTO dbo.Auftraege WITH (TABLOCK)

    SELECT TOP (@BatchSize) *

    FROM [Quell-Server].[Quell-DB].dbo.Auftraege

    WHERE AuftragsNr > @LetzterKey

    ORDER BY AuftragsNr;

    SET @Importiert = @@ROWCOUNT;

    SET @LetzterKey = @LetzterKey + @Importiert;

    -- Fortschritt sichern (Wiederaufnahme bei Fehler möglich)

    MERGE migration.Fortschritt AS f

    USING (SELECT 'Auftraege' AS T, @LetzterKey AS K,

                  @LetzterKey AS Z) AS s ON f.Tabelle = s.T

    WHEN MATCHED THEN UPDATE SET LetzterKey = s.K,

        MigriertZeilen = s.Z, Status = 'LAUFEND'

    WHEN NOT MATCHED THEN INSERT (Tabelle, LetzterKey, MigriertZeilen, Status)

        VALUES (s.T, s.K, s.Z, 'LAUFEND');

    PRINT 'Batch fertig: ' + CAST(@LetzterKey AS VARCHAR) + ' Zeilen gesamt';

    IF @Importiert = @BatchSize WAITFOR DELAY '00:00:01';

END

UPDATE migration.Fortschritt SET Status = 'ABGESCHLOSSEN' WHERE Tabelle = 'Auftraege';

-- Wenn Quell- und Zielschema unterschiedliche Strukturen haben

-- Mapping-Tabelle dokumentiert die Transformationslogik

CREATE TABLE migration.Feld_Mapping (

    MappingID           INT IDENTITY PRIMARY KEY,

    Quell_Tabelle       NVARCHAR(200),

    Quell_Feld          NVARCHAR(200),

    Quell_Typ           NVARCHAR(50),

    Ziel_Tabelle        NVARCHAR(200),

    Ziel_Feld           NVARCHAR(200),

    Ziel_Typ            NVARCHAR(50),

    Transformation      NVARCHAR(MAX),   -- SQL-Ausdruck

    Standardwert        NVARCHAR(200),   -- wenn Quelle NULL

    Validierungsregel   NVARCHAR(500),

    Kommentar           NVARCHAR(500),

    Getestet            BIT DEFAULT 0

);

-- Beispiel: ERP-Migration mit Schemaänderungen

INSERT INTO migration.Feld_Mapping

    (Quell_Tabelle, Quell_Feld, Ziel_Tabelle, Ziel_Feld,

     Transformation, Standardwert, Kommentar)

VALUES

('ALT_Kunden', 'KD_NR',    'Kunden', 'KundenNr',   'CAST(KD_NR AS INT)', NULL, 'War VARCHAR, jetzt INT'),

('ALT_Kunden', 'KD_NAME',  'Kunden', 'Name',        'TRIM(KD_NAME)',       NULL, 'Leerzeichen entfernen'),

('ALT_Kunden', 'KD_PLZ',   'Kunden', 'PLZ',         'RIGHT(''00000'' + KD_PLZ, 5)', NULL, 'PLZ auf 5 Stellen auffüllen'),

('ALT_Kunden', 'KD_LAND',  'Kunden', 'Land',        'ISNULL(KD_LAND, ''DE'')', '''DE''', 'NULL → DE'),

('ALT_Kunden', 'KD_AKTIV', 'Kunden', 'Gesperrt',    'CASE WHEN KD_AKTIV=1 THEN 0 ELSE 1 END', '0', 'Invertierte Logik!');

-- Transformations-View auf der Quelldatenbank

-- Kapselt alle Bereinigungen — Migration greift auf diese View zu

CREATE OR ALTER VIEW migration.V_Kunden_Transformiert AS

SELECT

    -- Typkonvertierungen

    CAST(KD_NR AS INT)                          AS KundenNr,

    TRIM(KD_NAME)                               AS Name,

    -- Normalisierung

    UPPER(TRIM(KD_NAME))                        AS Name_Norm,

    -- PLZ auffüllen und validieren

    CASE WHEN KD_PLZ LIKE '[0-9][0-9][0-9][0-9][0-9]'

         THEN KD_PLZ

         WHEN KD_PLZ LIKE '[0-9][0-9][0-9][0-9]'

         THEN '0' + KD_PLZ                      -- 4-stellige PLZ → 5-stellig

         ELSE NULL END                          AS PLZ,

    ISNULL(KD_LAND, 'DE')                       AS Land,

    -- Invertierte Logik

    CASE WHEN KD_AKTIV = 1 THEN 0 ELSE 1 END   AS Gesperrt,

    -- Datumskonvertierung (war VARCHAR im Altsystem)

    TRY_CONVERT(DATE, KD_ANLAGEDATUM, 104)      AS AnlageDatum, -- 104 = dd.mm.yyyy

    -- Datenqualitäts-Flag

    CASE

        WHEN KD_NR IS NULL THEN 'KEIN_KEY'

        WHEN TRIM(KD_NAME) = '' THEN 'KEIN_NAME'

        WHEN TRY_CONVERT(DATE, KD_ANLAGEDATUM, 104) IS NULL THEN 'DATUM_UNGUELTIG'

        ELSE 'OK'

    END                                         AS _Migrations_Status

FROM [Quell-Server].[ALT-DB].dbo.ALT_Kunden;

-- Migration via View: nur valide Zeilen übertragen

INSERT INTO dbo.Kunden (KundenNr, Name, PLZ, Land, Gesperrt, AnlageDatum)

SELECT KundenNr, Name, PLZ, Land, Gesperrt, AnlageDatum

FROM migration.V_Kunden_Transformiert

WHERE _Migrations_Status = 'OK';

-- Abgelehnte Zeilen in Quarantäne

INSERT INTO migration.Quarantaene (Quell_Tabelle, Quell_Key, Fehlergrund, Rohdaten)

SELECT 'ALT_Kunden', CAST(KundenNr AS NVARCHAR), _Migrations_Status,

       (SELECT * FROM migration.V_Kunden_Transformiert v2

        WHERE v2.KundenNr = v.KundenNr FOR JSON PATH)

FROM migration.V_Kunden_Transformiert v

WHERE _Migrations_Status <> 'OK';

-- Stufe 1: Row Count — grundlegende Vollständigkeit

CREATE OR ALTER PROCEDURE migration.SP_Validierung_RowCounts AS

BEGIN

    SELECT

        'Kunden'                    AS Tabelle,

        (SELECT COUNT(*) FROM [Quell-Server].[ALT-DB].dbo.ALT_Kunden

         WHERE KD_AKTIV = 1)        AS Quelle_Zeilen,

        (SELECT COUNT(*) FROM dbo.Kunden) AS Ziel_Zeilen,

        (SELECT COUNT(*) FROM migration.Quarantaene

         WHERE Quell_Tabelle = 'ALT_Kunden') AS Quarantaene_Zeilen,

        (SELECT COUNT(*) FROM [Quell-Server].[ALT-DB].dbo.ALT_Kunden

         WHERE KD_AKTIV = 1) -

        (SELECT COUNT(*) FROM dbo.Kunden) -

        (SELECT COUNT(*) FROM migration.Quarantaene

         WHERE Quell_Tabelle = 'ALT_Kunden')  AS Differenz,

        CASE WHEN (SELECT COUNT(*) FROM [Quell-Server].[ALT-DB].dbo.ALT_Kunden

                   WHERE KD_AKTIV = 1) =

                  (SELECT COUNT(*) FROM dbo.Kunden) +

                  (SELECT COUNT(*) FROM migration.Quarantaene

                   WHERE Quell_Tabelle = 'ALT_Kunden')

             THEN 'OK' ELSE 'FEHLER' END       AS Status

    UNION ALL

    SELECT 'Auftraege',

        (SELECT COUNT(*) FROM [Quell-Server].[ALT-DB].dbo.ALT_Auftraege),

        (SELECT COUNT(*) FROM dbo.Auftraege),

        (SELECT COUNT(*) FROM migration.Quarantaene

         WHERE Quell_Tabelle = 'ALT_Auftraege'),

        (SELECT COUNT(*) FROM [Quell-Server].[ALT-DB].dbo.ALT_Auftraege) -

        (SELECT COUNT(*) FROM dbo.Auftraege) -

        (SELECT COUNT(*) FROM migration.Quarantaene

         WHERE Quell_Tabelle = 'ALT_Auftraege'),

        CASE WHEN (SELECT COUNT(*) FROM [Quell-Server].[ALT-DB].dbo.ALT_Auftraege) =

                  (SELECT COUNT(*) FROM dbo.Auftraege) +

                  (SELECT COUNT(*) FROM migration.Quarantaene

                   WHERE Quell_Tabelle = 'ALT_Auftraege')

             THEN 'OK' ELSE 'FEHLER' END;

END

-- Stufe 2: Checksum — Datenintegrität auf Zeilenebene

CREATE OR ALTER PROCEDURE migration.SP_Validierung_Checksum AS

BEGIN

    -- Checksum über alle Zeilen und alle Spalten

    SELECT

        'Kunden'                    AS Tabelle,

        CHECKSUM_AGG(BINARY_CHECKSUM(*)) AS Quell_Checksum

    INTO #Quelle

    FROM [Quell-Server].[ALT-DB].dbo.ALT_Kunden WHERE KD_AKTIV = 1;

    SELECT

        'Kunden'                    AS Tabelle,

        CHECKSUM_AGG(BINARY_CHECKSUM(KundenNr, Name, PLZ, Land, AnlageDatum)) AS Ziel_Checksum

    INTO #Ziel FROM dbo.Kunden;

    SELECT q.Tabelle,

           q.Quell_Checksum, z.Ziel_Checksum,

           CASE WHEN q.Quell_Checksum = z.Ziel_Checksum THEN 'OK'

                ELSE 'CHECKSUM UNTERSCHIEDLICH' END AS Status

    FROM #Quelle q JOIN #Ziel z ON q.Tabelle = z.Tabelle;

END

-- Stufe 3: Business-Reconciliation — Summen und Kennzahlen

CREATE OR ALTER PROCEDURE migration.SP_Validierung_Business AS

BEGIN

    DECLARE @Tol DECIMAL(5,4) = 0.001; -- 0.1% Toleranz

    SELECT

        'Jahresumsatz ' + CAST(YEAR(GETDATE()) AS VARCHAR) AS Kennzahl,

        (SELECT SUM(AU_BETRAG) FROM [Quell-Server].[ALT-DB].dbo.ALT_Auftraege

         WHERE YEAR(AU_DATUM) = YEAR(GETDATE())) AS Quell_Wert,

        (SELECT SUM(Nettobetrag) FROM dbo.Auftraege

         WHERE YEAR(Auftragsdatum) = YEAR(GETDATE())) AS Ziel_Wert,

        ABS((SELECT SUM(Nettobetrag) FROM dbo.Auftraege

             WHERE YEAR(Auftragsdatum) = YEAR(GETDATE())) -

            (SELECT SUM(AU_BETRAG) FROM [Quell-Server].[ALT-DB].dbo.ALT_Auftraege

             WHERE YEAR(AU_DATUM) = YEAR(GETDATE()))) /

        NULLIF((SELECT SUM(AU_BETRAG) FROM [Quell-Server].[ALT-DB].dbo.ALT_Auftraege

                WHERE YEAR(AU_DATUM) = YEAR(GETDATE())), 0) AS Abweichung_Rel,

        CASE WHEN ABS((SELECT SUM(Nettobetrag) FROM dbo.Auftraege

                       WHERE YEAR(Auftragsdatum) = YEAR(GETDATE())) -

                      (SELECT SUM(AU_BETRAG) FROM [Quell-Server].[ALT-DB].dbo.ALT_Auftraege

                       WHERE YEAR(AU_DATUM) = YEAR(GETDATE()))) /

              NULLIF((SELECT SUM(AU_BETRAG) FROM [Quell-Server].[ALT-DB].dbo.ALT_Auftraege

                      WHERE YEAR(AU_DATUM) = YEAR(GETDATE())), 0) <= @Tol

             THEN 'OK' ELSE 'ABWEICHUNG > 0.1%' END AS Status;

END

-- Stufe 4: Referenzielle Integrität — keine verwaisten Fremdschlüssel

CREATE OR ALTER PROCEDURE migration.SP_Validierung_Integritaet AS

BEGIN

    -- Auftragspositionen ohne gültigen Auftrag

    SELECT 'Verwaiste Auftragspositionen' AS Pruefpunkt,

           COUNT(*) AS Anzahl,

           CASE WHEN COUNT(*) = 0 THEN 'OK' ELSE 'FEHLER' END AS Status

    FROM dbo.Auftragspositionen p

    WHERE NOT EXISTS (SELECT 1 FROM dbo.Auftraege a

                      WHERE a.AuftragsNr = p.AuftragsNr)

    UNION ALL

    SELECT 'Auftraege ohne gültigen Kunden',

           COUNT(*),

           CASE WHEN COUNT(*) = 0 THEN 'OK' ELSE 'FEHLER' END

    FROM dbo.Auftraege a

    WHERE NOT EXISTS (SELECT 1 FROM dbo.Kunden k

                      WHERE k.KundenNr = a.KundenNr);

END

-- Migrations-Abnahmeprotokoll: alle Stufen in einem Report

CREATE OR ALTER PROCEDURE migration.SP_Abnahmeprotokoll AS

BEGIN

    PRINT '=== MIGRATIONS-ABNAHMEPROTOKOLL ===';

    PRINT 'Datum: ' + CAST(GETDATE() AS NVARCHAR);

    PRINT '';

    -- Row Counts

    PRINT '--- STUFE 1: ROW COUNTS ---';

    EXEC migration.SP_Validierung_RowCounts;

    -- Checksums

    PRINT '--- STUFE 2: CHECKSUMS ---';

    EXEC migration.SP_Validierung_Checksum;

    -- Business-Reconciliation

    PRINT '--- STUFE 3: BUSINESS-RECONCILIATION ---';

    EXEC migration.SP_Validierung_Business;

    -- Referenzielle Integrität

    PRINT '--- STUFE 4: REFERENZIELLE INTEGRITAET ---';

    EXEC migration.SP_Validierung_Integritaet;

    -- Constraint-Status

    PRINT '--- CONSTRAINTS ---';

    SELECT Tabelle, Constraint_Name, Constraint_Typ, Status

    FROM (

        SELECT SCHEMA_NAME(t.schema_id)+'.'+t.name AS Tabelle,

               c.name AS Constraint_Name, 'CHECK' AS Constraint_Typ,

               CASE WHEN c.is_disabled=1 THEN 'DEAKTIVIERT!'

                    WHEN c.is_not_trusted=1 THEN 'NICHT VERIFIZIERT!'

                    ELSE 'OK' END AS Status

        FROM sys.check_constraints c JOIN sys.tables t ON c.parent_object_id=t.object_id

        UNION ALL

        SELECT SCHEMA_NAME(t.schema_id)+'.'+t.name,f.name,'FOREIGN KEY',

               CASE WHEN f.is_disabled=1 THEN 'DEAKTIVIERT!'

                    WHEN f.is_not_trusted=1 THEN 'NICHT VERIFIZIERT!'

                    ELSE 'OK' END

        FROM sys.foreign_keys f JOIN sys.tables t ON f.parent_object_id=t.object_id

    ) c WHERE Status <> 'OK';

    -- Quarantäne-Zusammenfassung

    PRINT '--- QUARANTAENE ---';

    SELECT Quell_Tabelle, Fehlergrund, COUNT(*) AS Anzahl

    FROM migration.Quarantaene

    GROUP BY Quell_Tabelle, Fehlergrund

    ORDER BY Anzahl DESC;

END

-- Cutover-Protokoll: jeder Schritt wird dokumentiert

CREATE TABLE migration.Cutover_Protokoll (

    SchrittID       INT IDENTITY PRIMARY KEY,

    Zeitpunkt       DATETIME2 DEFAULT SYSUTCDATETIME(),

    Schritt_Nr      INT,

    Schritt_Name    NVARCHAR(200),

    Ausfuehrender   NVARCHAR(100),

    Status          NVARCHAR(20),   -- GEPLANT / LAUFEND / ABGESCHLOSSEN / FEHLER

    Dauer_Min       DECIMAL(8,2),

    Kommentar       NVARCHAR(500)

);

-- Cutover-Schritte vordefinieren

INSERT INTO migration.Cutover_Protokoll

    (Schritt_Nr, Schritt_Name, Ausfuehrender, Status)

VALUES

(1,  'Backup Quell-System erstellen und verifizieren',  'DBA', 'GEPLANT'),

(2,  'Quell-System für neue Buchungen sperren',          'App-Team', 'GEPLANT'),

(3,  'Delta-Migration: alle Änderungen seit letzter Generalprobe', 'DBA', 'GEPLANT'),

(4,  'Row Count Validierung aller Tabellen',             'DBA', 'GEPLANT'),

(5,  'Business-Reconciliation ausführen',                'Controlling', 'GEPLANT'),

(6,  'Foreign Keys und Constraints aktivieren',          'DBA', 'GEPLANT'),

(7,  'Logins und Berechtigungen prüfen',                 'DBA', 'GEPLANT'),

(8,  'SQL Agent Jobs aktivieren',                        'DBA', 'GEPLANT'),

(9,  'Anwendungs-Verbindungsstring auf neues System',    'App-Team', 'GEPLANT'),

(10, 'Smoke Test: kritische Funktionen prüfen',          'Fachbereich', 'GEPLANT'),

(11, 'Go/No-Go Entscheidung',                            'Projektleitung', 'GEPLANT'),

(12, 'Monitoring aktivieren',                            'DBA', 'GEPLANT');

-- Rollback in unter 30 Minuten: Verbindungen zurück auf Altsystem

-- Schritt 1: Anwendung auf Altsystem-Verbindungsstring zurückschalten

-- Schritt 2: Neues System für neue Buchungen sperren

-- Schritt 3: Delta-Daten aus neuem System in Altsystem zurückspielen (wenn nötig)

-- Rollback-Entscheidungsmatrix

CREATE TABLE migration.Rollback_Trigger (

    TriggerID       INT IDENTITY PRIMARY KEY,

    Bedingung       NVARCHAR(500),   -- Was löst Rollback aus?

    Verantwortlich  NVARCHAR(100),   -- Wer entscheidet?

    Max_Reaktionszeit_Min INT        -- Wie schnell muss entschieden werden?

);

INSERT INTO migration.Rollback_Trigger (Bedingung, Verantwortlich, Max_Reaktionszeit_Min)

VALUES

('Row Count Validierung zeigt > 0.1% Abweichung', 'DBA + Projektleitung', 15),

('Business-Reconciliation Abweichung > 0.1%', 'Controller + Projektleitung', 15),

('Foreign Key Aktivierung schlägt fehl', 'DBA', 10),

('Anwendungs-Smoke-Test schlägt fehl', 'App-Team + Projektleitung', 20),

('Kritische Anwendungsfehler > 30 Minuten nach Go-Live', 'Projektleitung', 30),

('Datenverlust bei spezifischen Datensätzen bestätigt', 'Projektleitung', 5);

VOR DEM START

Definieren Sie das Rollback-Kriterium schriftlich, bevor die erste Zeile Code geschrieben wird. Benennen Sie den Entscheidungsträger für den Go/No-Go-Moment. Erstellen Sie ein vollständiges Backup des Quellsystems und verifizieren Sie die Wiederherstellbarkeit. Richten Sie die Testumgebung mit einer vollständigen Kopie der Produktivdaten ein.

ERGEBNIS NACH 30 TAGEN

Ihr Ergebnis nach 30 Tagen: Eine vollständig migrierte, validierte und in Betrieb genommene Datenbank mit dokumentiertem Abnahmeprotokoll, aktivierten Constraints, konfigurierten Jobs, verifizierten Logins und einem laufenden Monitoring-Dashboard. Das Altsystem bleibt noch vier Wochen im Read-only-Betrieb als Rückfalloption.

SQL-Migrations-Begleitung

Planung, Durchführung und Validierung von SQL Server-Datenmigrationsprojekten — von der Quellenanalyse bis zur Go-Live-Abnahme.

ERP-Einführungsberatung

Begleitung von ERP-Projekten (MACH, Dynamics NAV, APplus) — Datenmigration, Transformation, Validierung und Go-Live-Support.

SQL Server DB Health Check

Professioneller Audit — Performance, Migrationsbereitschaft, Kompatibilitäts-Level, Backup-Strategie. Scope: 3-5 Tage.

Datenbankdesign & Refactoring

Strukturiertes Refactoring und Schema-Modernisierung als Vorbereitung für Migrationen.

Interim IT-Management

Übernahme der IT-Steuerung auf Zeit — Projektleitung, Dienstleister-Management, strategische IT-Ausrichtung.