Ein Anwendungsbeispiel für die neue Merge-Funktion in PostgreSQL 15
Mit dem PostgreSQL Release Version 15 wurde die längst überfällige Merge-Funktion eingeführt, wodurch ein potenzieller Umstieg von Oracle auf PostgreSQL noch einfacher gestaltet werden kann – hierzu ein Anwendungsbeispiel für die Merge-Funktion.
Die Anwendungslogik
Mit unserer Beispielanwendung wollen wir simple Informationen zu einem Datenbanksystem in einer Tabelle speichern, Änderungen protokollieren und hierzu ein „Staging“-Verfahren anwenden. Wir erstellen dazu die folgenden drei Tabellen:
create table db_info ( db_id int not null, db_version varchar(30), db_ip varchar(20), last_found timestamp ); create table db_info_hist ( db_id int not null, db_version varchar(30), db_ip varchar(20), change_date timestamp, action char(1) ); create table db_info_stage ( db_id int not null, db_version varchar(30), db_ip varchar(20) );
Neben einer Datenbank-ID wollen wir auch die Version und IP-Adresse der Datenbank speichern. Für unser Beispiel gehen wir davon aus, dass die Informationen täglich durch einen Job gesammelt und anschließend als einzelne Datensätze in die Tabelle DB_INFO_STAGE geschrieben werden.
Die Magie des Merges
Für einen Abgleich, ob es sich bei den Datensätzen aus der „Staging“-Tabelle um neue oder bereits bekannte Datensätze handelt, nutzen wir Merge innerhalb einer Prozedur:
create or replace procedure prc_merge_db_info() as $$
begin
merge into db_info t
using (
select db_id, db_version, db_ip from db_info_stage
except
select db_id, db_version, db_ip from db_info
) s
on t.db_id = s.db_id
when matched then
update set
db_version = s.db_version,
db_ip = s.db_ip
when not matched then
insert (db_id, db_version, db_ip, last_found) values (s.db_id, s.db_version, s.db_ip, current_timestamp);
update db_info set last_found = current_timestamp where db_id in (select db_id from db_info_stage);
delete from db_info_stage;
end;
$$
language plpgsql
Zunächst definieren wir innerhalb des Merges eine „Zielmenge“, welche mit der Zieltabelle DB_INFO abgeglichen werden soll:
…
using (
select db_id, db_version, db_ip from db_info_stage
except
select db_id, db_version, db_ip from db_info
) s
…
Die Datenmenge „s“ enthält alle Datensätze der „Staging“-Tabelle abzüglich der bereits vorhandenen Datensätze aus der Zieltabelle DB_INFO. Anhand der DB_ID prüfen wir nun Datensatz für Datensatz, ob es sich um einen neuen oder veränderten Datensatz handelt:
… on t.db_id = s.db_id …
Trifft die "ON"-Klausel zu, existiert bereits ein Datensatz für die jeweilige Datenbank. Somit muss es sich in diesem Fall um eine veränderte Information handeln. In unserem Beispiel kann sich z. B. die IP-Adresse der Datenbank verändert haben.
Das nachfolgende UPDATE-Statement passt den Datensatz in der Zieltabelle entsprechend an:
…
when matched then
update set
db_version = s.db_version,
db_ip = s.db_ip
…
Stimmt die „ON“-Klausel nicht überein, handelt es sich um einen neuen Datensatz. Dementsprechend wird bei „not matched“ ein INSERT in die Zieltabelle ausgeführt:
…
when not matched then
insert (db_id, db_version, db_ip, last_found) values (s.db_id, s.db_version, s.db_ip, current_timestamp);
…
Bevor die "Staging"-Tabelle abschließend wieder geleert wird, führen wir ein UPDATE auf alle übereinstimmenden DB_IDs durch und aktualisieren die „LAST_FOUND“-Spalte:
… update db_info set last_found = current_timestamp where db_id in (select db_id from db_info_stage); …
Der History-Trigger
Damit wir auch die historischen Informationen speichern können, ob sich ein Datensatz in der Zieltabelle mit der Zeit verändert hat, dieser hinzugefügt oder letztlich gelöscht wurde, benötigen wir noch einen Trigger für die Zieltabelle:
create or replace function trg_db_info_staging()
returns trigger as
$func$
begin
if (TG_OP = 'UPDATE') THEN
if (new.db_version != old.db_version) or (new.db_ip != old.db_ip) then
insert into db_info_hist(db_id, db_version, db_ip, change_date, action) values (new.db_id, new.db_version, new.db_ip, current_timestamp, 'U');
end if;
elsif (TG_OP = 'INSERT') THEN
insert into db_info_hist(db_id, db_version, db_ip, change_date, action) values (new.db_id, new.db_version, new.db_ip, current_timestamp, 'I');
else
insert into db_info_hist(db_id, change_date, action) values (old.db_id, current_timestamp, 'D');
end if;
return new;
end
$func$ language plpgsql;
create or replace trigger db_info_staging_trg
after insert or update or delete on db_info
for each row
execute function trg_db_info_staging();
/
Der Trigger sorgt final dafür, dass wir in der DB_INFO_HIST-Tabelle ein Aktionsfeld mit einem I (= INSERT), D (= DELETE) oder U (= UPDATE) gefüllt bekommen. Zusammen mit dem Timestamp der Aktion lässt sich so der genaue „Zustand“ der Zieltabelle zu einem jedem beliebigen Zeitpunkt abbilden.
Der Merge in Aktion
Für unser Beispiel fügen wir zunächst vier simple Datensätze mit den DB_IDs 100, 200, 300 und 400 in die „Staging“-Tabelle ein und führen die Merge-Prozedur aus:
insert into db_info_stage values(100,'19.10.0.0','192.168.56.101'); insert into db_info_stage values(200,'19.10.0.0','192.168.56.102'); insert into db_info_stage values(300,'19.10.0.0','192.168.56.103'); insert into db_info_stage values(400,'19.10.0.0','192.168.56.104'); call prc_merge_db_info();
Bei einem Blick auf die Ziel- und History-Tabelle sehen wir nun Folgendes:
db_id | db_version | db_ip | last_found -------+------------+----------------+--------------------- 400 | 19.10.0.0 | 192.168.56.104 | 10.01.2023 09:38:46 100 | 19.10.0.0 | 192.168.56.101 | 10.01.2023 09:38:46 300 | 19.10.0.0 | 192.168.56.103 | 10.01.2023 09:38:46 200 | 19.10.0.0 | 192.168.56.102 | 10.01.2023 09:38:46 (4 rows) db_id | db_version | db_ip | change_date | action -------+------------+----------------+---------------------+-------- 100 | 19.10.0.0 | 192.168.56.101 | 10.01.2023 09:38:46 | I 200 | 19.10.0.0 | 192.168.56.102 | 10.01.2023 09:38:46 | I 300 | 19.10.0.0 | 192.168.56.103 | 10.01.2023 09:38:46 | I 400 | 19.10.0.0 | 192.168.56.104 | 10.01.2023 09:38:46 | I (4 rows)
Jeder neue Datensatz wird mit einem „I“ in die History-Tabelle geschrieben. Alle Datensätze erhalten den aktuellen Timestamp als „LAST_FOUND“ bzw. „CHANGE_DATE“ gespeichert.
Nun simulieren wir einen späteren Lauf. Hier ändern wir bewusst die IP-Adressen von DB_ID 200 und 300:
insert into db_info_stage values(100,'19.10.0.0','192.168.56.101'); insert into db_info_stage values(200,'19.10.0.0','192.168.56.202'); insert into db_info_stage values(300,'19.10.0.0','192.168.56.303'); insert into db_info_stage values(400,'19.10.0.0','192.168.56.104'); call prc_merge_db_info(); db_id | db_version | db_ip | last_found -------+------------+----------------+--------------------- 400 | 19.10.0.0 | 192.168.56.104 | 10.01.2023 09:41:05 100 | 19.10.0.0 | 192.168.56.101 | 10.01.2023 09:41:05 300 | 19.10.0.0 | 192.168.56.303 | 10.01.2023 09:41:05 200 | 19.10.0.0 | 192.168.56.202 | 10.01.2023 09:41:05 (4 rows) db_id | db_version | db_ip | change_date | action -------+------------+----------------+---------------------+-------- 100 | 19.10.0.0 | 192.168.56.101 | 10.01.2023 09:38:46 | I 200 | 19.10.0.0 | 192.168.56.102 | 10.01.2023 09:38:46 | I 200 | 19.10.0.0 | 192.168.56.202 | 10.01.2023 09:41:05 | U 300 | 19.10.0.0 | 192.168.56.103 | 10.01.2023 09:38:46 | I 300 | 19.10.0.0 | 192.168.56.303 | 10.01.2023 09:41:05 | U 400 | 19.10.0.0 | 192.168.56.104 | 10.01.2023 09:38:46 | I (6 rows)
Während die Ziel-Tabelle weiterhin aus den vier vorhandenen Datensätzen mit den aktuellen IP-Adressen und Timestamps besteht, wurde die History-Tabelle um die Update-Informationen zzgl. der neuen IP-Adresse erweitert.
Fazit
Die neue PostgreSQL Merge-Funktion steht den Oracle Funktionalitäten in keiner Weise nach. Die Syntax in diesem Beispiel ist bis auf 2-3 Kleinigkeiten identisch (siehe Oracle-Code unten 🠗). In Kombination mit einer zusätzlichen Trigger-Logik lässt sich so eine komplexe Staging- und History-Logik implementieren.
In meinem konkreten Beispiel habe ich die Logik aus einer Oracle-Anwendung mit wenig Aufwand auf PostgreSQL adaptiert.
Haben auch Sie einen Anwendungsfall für die Merge-Funktion oder benötigen Sie grundsätzlich Hilfe in Sachen PostgreSQL- und Oracle-Entwicklungsthemen? Sprechen Sie uns einfach an.
create or replace procedure prc_merge_db_info
as
begin
merge into db_info t
using (
select db_id, db_version, db_ip from db_info_stage
minus
select db_id, db_version, db_ip from db_info
) s
on (t.db_id = s.db_id)
when matched then
update set
t.db_version = s.db_version,
t.db_ip = s.db_ip
when not matched then
insert (db_id, db_version, db_ip, last_found) values (s.db_id, s.db_version, s.db_ip, sysdate);
update db_info set last_found = sysdate where db_id in (select db_id from db_info_stage);
delete from db_info_stage;
end;
/
Seminarempfehlung
POSTGRESQL ADMINISTRATION DB-PG-01
Zum Seminar
Bei Updates im Blog, informieren wir per E-Mail.
Kommentare