granicus.if.org Git - postgresql/blob - contrib/test_decoding/sql/replorigin.sql

   1 -- predictability
   2 SET synchronous_commit = on;
   3
   4 CREATE TABLE origin_tbl(id serial primary key, data text);
   5 CREATE TABLE target_tbl(id serial primary key, data text);
   6
   7 SELECT pg_replication_origin_create('test_decoding: regression_slot');
   8 -- ensure duplicate creations fail
   9 SELECT pg_replication_origin_create('test_decoding: regression_slot');
  10
  11 --ensure deletions work (once)
  12 SELECT pg_replication_origin_create('test_decoding: temp');
  13 SELECT pg_replication_origin_drop('test_decoding: temp');
  14 SELECT pg_replication_origin_drop('test_decoding: temp');
  15
  16 SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
  17
  18 -- origin tx
  19 INSERT INTO origin_tbl(data) VALUES ('will be replicated and decoded and decoded again');
  20 INSERT INTO target_tbl(data)
  21 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
  22
  23 -- as is normal, the insert into target_tbl shows up
  24 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
  25
  26 INSERT INTO origin_tbl(data) VALUES ('will be replicated, but not decoded again');
  27
  28 -- mark session as replaying
  29 SELECT pg_replication_origin_session_setup('test_decoding: regression_slot');
  30
  31 -- ensure we prevent duplicate setup
  32 SELECT pg_replication_origin_session_setup('test_decoding: regression_slot');
  33
  34 BEGIN;
  35 -- setup transaction origin
  36 SELECT pg_replication_origin_xact_setup('0/aabbccdd', '2013-01-01 00:00');
  37 INSERT INTO target_tbl(data)
  38 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'only-local', '1');
  39 COMMIT;
  40
  41 -- check replication progress for the session is correct
  42 SELECT pg_replication_origin_session_progress(false);
  43 SELECT pg_replication_origin_session_progress(true);
  44
  45 SELECT pg_replication_origin_session_reset();
  46
  47 SELECT local_id, external_id, remote_lsn, local_lsn <> '0/0' FROM pg_replication_origin_status;
  48
  49 -- check replication progress identified by name is correct
  50 SELECT pg_replication_origin_progress('test_decoding: regression_slot', false);
  51 SELECT pg_replication_origin_progress('test_decoding: regression_slot', true);
  52
  53 -- ensure reset requires previously setup state
  54 SELECT pg_replication_origin_session_reset();
  55
  56 -- and magically the replayed xact will be filtered!
  57 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'only-local', '1');
  58
  59 --but new original changes still show up
  60 INSERT INTO origin_tbl(data) VALUES ('will be replicated');
  61 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1',  'only-local', '1');
  62
  63 SELECT pg_drop_replication_slot('regression_slot');
  64 SELECT pg_replication_origin_drop('test_decoding: regression_slot');