CREATE OR REPLACE PACKAGE BODY vergleich_10g
AS
-- lokale Variablen raus
    v_col_str        VARCHAR2(3000);
    stmt             VARCHAR2(3000);
    v_schema_remote  VARCHAR2(30);
    TYPE spalten_type IS TABLE OF VARCHAR2(2000) INDEX BY BINARY_INTEGER;
    spalten_tab   spalten_type;

/* Die private Funktion Spalten stellt die Spaltenliste entweder aus allen
   oder den vorgegebenen Spalten (die Primärschlüsselspalte muss dabei sein) zusammen */
FUNCTION spalten (
    p_schema        VARCHAR2,
    p_tabelle       VARCHAR2,
    p_link          VARCHAR2 DEFAULT NULL,
    p_spaltepk      VARCHAR2 DEFAULT NULL,
    p_spalte1       VARCHAR2 DEFAULT NULL,
    p_spalte2       VARCHAR2 DEFAULT NULL,
    p_spalte3       VARCHAR2 DEFAULT NULL) RETURN VARCHAR2
IS
BEGIN
-- die angegebenen Spalten werden konkateniert,
-- TRIM entfernt überschüssige Pipes von außen, REPLACE von innen (falls z.B. p_spalte1 ausgelassen wurde)
    v_col_str := REPLACE(TRIM('|' FROM p_spaltepk||'||'||p_spalte1||'||'||p_spalte2||'||'||p_spalte3), '||||', '||');
  -- Falls keine Spalten angegeben wurden, werden die Spalten der Tabelle aus dba_tab_columns ausgelesen
  -- und in das Array spalten_tab verfrachtet
    IF v_col_str IS NULL THEN
      -- wenn ein DB-Link angegeben ist, muss dieser Select dynamisch erstellt werden
       IF p_link IS NOT NULL THEN
          stmt := 'SELECT column_name FROM dba_tab_columns@'||p_link||
                  q'# WHERE table_name = UPPER('#'||p_tabelle||q'#') AND owner = UPPER('#'||p_schema||q'#') ORDER BY column_name#';
          EXECUTE IMMEDIATE stmt BULK COLLECT INTO spalten_tab;
       ELSE
          SELECT column_name  BULK COLLECT INTO spalten_tab FROM dba_tab_columns
          WHERE table_name = UPPER(p_tabelle) AND owner = UPPER(p_schema)
          ORDER BY column_name;
       END IF;
       FOR i IN spalten_tab.FIRST .. spalten_tab.LAST LOOP
            v_col_str := v_col_str||'||'||spalten_tab(i);
       END LOOP;
     END IF;
     v_col_str := LTRIM(v_col_str, '|');
     RETURN v_col_str;
EXCEPTION
  WHEN VALUE_ERROR THEN
   DBMS_OUTPUT.PUT_LINE('Bitte Eingaben überprüfen (Tabellen-, DB-Link-, Schema-Namen etc.)');
  RAISE;
  WHEN OTHERS THEN
    RAISE;
END;

/* Die private Funktion hash_sum ermittelt den Hash-Wert für alle oder bis zu 4
   explizit angegebene Spalten einer Tabelle*/
FUNCTION hash_sum (
    p_schema        VARCHAR2,
    p_tabelle       VARCHAR2,
    p_link          VARCHAR2 DEFAULT NULL,
    p_spaltepk      VARCHAR2 DEFAULT NULL,
    p_spalte1       VARCHAR2 DEFAULT NULL,
    p_spalte2       VARCHAR2 DEFAULT NULL,
    p_spalte3       VARCHAR2 DEFAULT NULL) RETURN NUMBER
IS
    hash_wert NUMBER := 0;
BEGIN
   	   v_col_str := spalten(
        p_schema   => p_schema,
        p_tabelle  => p_tabelle,
        p_link     => p_link,
        p_spaltepk => p_spaltepk,
        p_spalte1  => p_spalte1,
        p_spalte2  => p_spalte2,
        p_spalte3  => p_spalte3);
		 -- wenn ein DB-Link angegeben ist
stmt := 'SELECT SUM(ORA_HASH('||v_col_str||',4294967295)) FROM '||p_schema||'.'||p_tabelle;
     IF p_link IS NOT NULL THEN
        stmt := stmt ||'@'||p_link;
     END IF;
      EXECUTE IMMEDIATE stmt INTO hash_wert;
      RETURN hash_wert;
EXCEPTION
  WHEN OTHERS THEN
   RAISE;
END;

/* Die Funktion vergl_hash vergleicht die Hash-Werte der beiden Tabellen - per default für alle Spalten -
   für eine schnelle Information, ob die Objekte gleich oder verschieden sind.  */
PROCEDURE vergl_hash (
     p_schema_lokal      VARCHAR2,
     p_basistabelle      VARCHAR2,
     p_schema_remote     VARCHAR2 DEFAULT NULL,
     p_vergleichstabelle VARCHAR2,
     p_link              VARCHAR2 DEFAULT NULL,
     p_spaltepk          VARCHAR2 DEFAULT NULL,
     p_spalte1           VARCHAR2 DEFAULT NULL,
     p_spalte2           VARCHAR2 DEFAULT NULL,
     p_spalte3           VARCHAR2 DEFAULT NULL)
IS
   v_vergleichstabelle  VARCHAR2(30);
   v_host    VARCHAR2(30);
   n_hash_1  NUMBER := 0;
   n_hash_2  NUMBER := 0;
BEGIN
-- wenn der Name des Remote-Schemas nicht angegeben wurde (p_schema_remote ist per default NULL ),
-- wird der Name des lokalen Schemas genommen
    v_schema_remote := NVL(p_schema_remote, p_schema_lokal);
	-- hier wird der Name der über einen DB-Link verbundenen DB für die Textausgabe unten bestimmt
    IF p_link IS NOT NULL THEN
       stmt := q'#SELECT host FROM all_db_links WHERE db_link = UPPER('#'||p_link||q'#')#';
       EXECUTE IMMEDIATE stmt INTO v_host ;
    END IF;
      v_vergleichstabelle := RTRIM(p_vergleichstabelle||'@'||p_link, '@');
/* n_hash_1 ist der Hash-Wert der Basistabelle. Die Basistabelle muss sich in einem
   Schema der lokalen Datenbank befinden */
     n_hash_1 := hash_sum (
         p_schema => p_schema_lokal,
        p_tabelle => p_basistabelle,
       p_spaltepk => p_spaltepk,
        p_spalte1 => p_spalte1,
        p_spalte2 => p_spalte2,
        p_spalte3 => p_spalte3);
     DBMS_OUTPUT.PUT_LINE('Hash-Wert der Basistabelle '||p_schema_lokal||'.'||p_basistabelle||': '||n_hash_1);
/* n_hash_2 ist der Hash-Wert der Vergleichstabelle */
        n_hash_2 := hash_sum (
         p_schema => v_schema_remote,
        p_tabelle => p_vergleichstabelle,
        p_link    => p_link,
       p_spaltepk => p_spaltepk,
        p_spalte1 => p_spalte1,
        p_spalte2 => p_spalte2,
        p_spalte3 => p_spalte3);
    DBMS_OUTPUT.PUT_LINE('Hash-Wert der Vergleichstabelle '||v_schema_remote||'.'||RTRIM(p_vergleichstabelle||'@'||v_host, '@')||': '||n_hash_2);
         IF n_hash_1 = n_hash_2 THEN
        DBMS_OUTPUT.PUT_LINE ('Keine Unterschiede gefunden');
    ELSE
        DBMS_OUTPUT.PUT_LINE('Die Tabellen stimmen nicht überein');
    END IF;
EXCEPTION
 WHEN OTHERS THEN
      RAISE;
 END vergl_hash;

/* Die Prozedur vergl_diff trägt die Differenzen, hier für bis zu 4 Spalten, in eine dynamisch erstellte Tabelle diff ein.*/
PROCEDURE vergl_diff(
     p_schema_lokal      VARCHAR2,
     p_basistabelle      VARCHAR2,
     p_schema_remote     VARCHAR2 DEFAULT NULL,
     p_vergleichstabelle VARCHAR2,
     p_link              VARCHAR2 DEFAULT NULL,
     p_spaltepk          VARCHAR2,
     p_spalte1           VARCHAR2,
     p_spalte2           VARCHAR2,
     p_spalte3           VARCHAR2,
     p_sync              BOOLEAN DEFAULT FALSE)
IS
  v_vergleichstabelle  VARCHAR2(60);
  v_count              NUMBER;
BEGIN
    v_schema_remote := NVL(p_schema_remote, p_schema_lokal);
    v_vergleichstabelle := RTRIM(p_vergleichstabelle||'@'||p_link, '@');
    v_col_str := REPLACE (spalten(
        p_schema   => p_schema_lokal,
        p_tabelle  => p_basistabelle,
        p_link     => p_link,
        p_spaltepk => p_spaltepk,
        p_spalte1  => p_spalte1,
        p_spalte2  => p_spalte2,
        p_spalte3  => p_spalte3), '||', ',');
/* Das Löschen der Tabelle diff im Subblock mit Fehlerbehandlung verhindert, dass die Prozedur abbricht, wenn die Tabelle diff noch nicht existiert. */
     BEGIN
         EXECUTE IMMEDIATE 'DROP TABLE diff';
     EXCEPTION
         WHEN OTHERS THEN NULL;
     END;
      EXECUTE IMMEDIATE 'CREATE TABLE diff AS SELECT '||v_col_str||' FROM '||p_schema_lokal||'.'||p_basistabelle||' WHERE 1=2';
      EXECUTE IMMEDIATE 'ALTER TABLE diff ADD(in_basis VARCHAR2(20), in_vergleich VARCHAR2(20))';
     stmt := 'INSERT INTO diff SELECT '||v_col_str||
     q'#, CASE COUNT(src1) WHEN 0 THEN 'fehlt' ELSE 'vorhanden' END in_basis,
          CASE COUNT(src2) WHEN 0 THEN 'fehlt' ELSE 'vorhanden' END in_vergleich
        FROM (SELECT #'||v_col_str||', 1 src1, TO_NUMBER(null) src2 FROM '||p_schema_lokal||'.'||p_basistabelle||' UNION ALL SELECT '
                       ||v_col_str||', TO_NUMBER(null) src1, 2 src2 FROM '||v_schema_remote||'.'||v_vergleichstabelle
                       ||') GROUP BY '||v_col_str||' HAVING COUNT(src1) <> COUNT(src2) ORDER BY 1';
     EXECUTE IMMEDIATE stmt;
     COMMIT;
      stmt := 'SELECT COUNT(*) FROM DIFF';
      EXECUTE IMMEDIATE stmt INTO v_count;
      DBMS_OUTPUT.PUT_LINE('Es wurden '||NVL(v_count, 0)||' abweichende Datensätze gefunden und in die Tabelle Diff eingetragen');
    IF p_sync THEN
       sync(
          p_schema_lokal  => p_schema_lokal,
          p_basistabelle  => p_basistabelle,
          p_schema_remote => v_schema_remote,
      p_vergleichstabelle => p_vergleichstabelle,
                   p_link => p_link,
               p_spaltepk => p_spaltepk,
               p_spalte1  => p_spalte1,
               p_spalte2  => p_spalte2,
               p_spalte3  => p_spalte3);
    END IF;
EXCEPTION
   WHEN OTHERS THEN
      RAISE;
END;

/* Die Prozedur sync synchronisiert die Tabellen über einen Merge*/
PROCEDURE sync (
     p_schema_lokal      VARCHAR2,
     p_basistabelle      VARCHAR2,
     p_schema_remote     VARCHAR2 DEFAULT NULL,
     p_vergleichstabelle VARCHAR2,
     p_link              VARCHAR2 DEFAULT NULL,
     p_spaltepk          VARCHAR2,
     p_spalte1           VARCHAR2,
     p_spalte2           VARCHAR2,
     p_spalte3           VARCHAR2)
IS
    v_vergleichstabelle  VARCHAR2(60);
BEGIN
   v_schema_remote := NVL(p_schema_remote, p_schema_lokal);
   v_vergleichstabelle := RTRIM(p_vergleichstabelle||'@'||p_link, '@');
  /* Wenn ein DB-Link angegeben wurde (in diesem Fall macht der Merge-Befehl Probleme), dann werden alle Datensätze,
     für die in der Tabelle diff in der Spalte in_vergleich 'vorhanden' steht, aus der Vergleichstabelle gelöscht
     und die entsprechenden Zeilen mit dem Eintrag in_basis = vorhanden in die Vergleichstabelle insertiert */
      DBMS_OUTPUT.PUT_LINE('Synchronisation gestartet');
      IF p_link IS NOT NULL THEN
        stmt := 'DELETE FROM '||v_schema_remote||'.'||v_vergleichstabelle
           ||' WHERE '||p_spaltepk||' IN (SELECT '||p_spaltepk||q'# FROM diff WHERE in_vergleich = 'vorhanden')#';
        EXECUTE IMMEDIATE stmt;
        stmt := 'INSERT INTO '||v_schema_remote||'.'||v_vergleichstabelle||' SELECT * FROM '||p_schema_lokal||'.'||p_basistabelle
           ||' WHERE '||p_spaltepk||' IN (SELECT '||p_spaltepk||q'# FROM diff WHERE in_basis = 'vorhanden')#';
      EXECUTE IMMEDIATE stmt;
      ELSE
/* Wenn die Vergleichstabelle in derselben DB liegt, kann man gezielt die angegebenen Spalten in der Vergleichstabelle
          updaten, die überflüssigen Zeilen entfernen und die fehlenden ergänzen */
        stmt := 'MERGE INTO '||v_schema_remote||'.'||v_vergleichstabelle||' o USING diff d '
           ||' ON (o.'||p_spaltepk||' = d.'||p_spaltepk||')'
           ||' WHEN MATCHED THEN UPDATE SET o.'||p_spalte1||' = d.'||p_spalte1||', o.'||p_spalte2||' = d.'||p_spalte2
           ||', o.'||p_spalte3||' = d.'||p_spalte3
           ||' WHERE o.'||p_spaltepk||' IN (SELECT '||p_spaltepk||' FROM diff GROUP BY '||p_spaltepk||' HAVING COUNT('||p_spaltepk||') =2)'
           ||q'# AND d.in_basis = 'vorhanden'#';
        EXECUTE IMMEDIATE stmt;
       stmt := 'DELETE FROM '||v_schema_remote||'.'||v_vergleichstabelle
           ||' WHERE '||p_spaltepk||' IN (SELECT '||p_spaltepk||q'# FROM diff WHERE in_vergleich = 'vorhanden'#'
           ||' MINUS SELECT '||p_spaltepk||' FROM diff GROUP BY '||p_spaltepk||' HAVING COUNT('||p_spaltepk||') =2)';
       EXECUTE IMMEDIATE stmt;
       stmt := 'INSERT INTO '||v_schema_remote||'.'||v_vergleichstabelle||' SELECT * FROM '||p_schema_lokal||'.'||p_basistabelle
           ||' WHERE '||p_spaltepk||' IN (SELECT '||p_spaltepk||q'# FROM diff WHERE in_vergleich = 'fehlt'#'
           ||' MINUS SELECT '||p_spaltepk||' FROM diff GROUP BY '||p_spaltepk||' HAVING COUNT('||p_spaltepk||') =2)';
       EXECUTE IMMEDIATE stmt;
     END IF;
     DBMS_OUTPUT.PUT_LINE('Synchronisation abgeschlossen');
    COMMIT;
EXCEPTION
   WHEN OTHERS THEN
      RAISE;
END sync;

END vergleich_10g;