contrib/test_decoding/sql/replorigin.sql

   1 -- predictability
   2 SET synchronous_commit = on;
   3
   4 -- superuser required by default
   5 CREATE ROLE regress_origin_replication REPLICATION;
   6 SET ROLE regress_origin_replication;
   7 SELECT pg_replication_origin_advance('regress_test_decoding: perm', '0/1');
   8 SELECT pg_replication_origin_create('regress_test_decoding: perm');
   9 SELECT pg_replication_origin_drop('regress_test_decoding: perm');
  10 SELECT pg_replication_origin_oid('regress_test_decoding: perm');
  11 SELECT pg_replication_origin_progress('regress_test_decoding: perm', false);
  12 SELECT pg_replication_origin_session_is_setup();
  13 SELECT pg_replication_origin_session_progress(false);
  14 SELECT pg_replication_origin_session_reset();
  15 SELECT pg_replication_origin_session_setup('regress_test_decoding: perm');
  16 SELECT pg_replication_origin_xact_reset();
  17 SELECT pg_replication_origin_xact_setup('0/1', '2013-01-01 00:00');
  18 SELECT pg_show_replication_origin_status();
  19 RESET ROLE;
  20 DROP ROLE regress_origin_replication;
  21
  22 CREATE TABLE origin_tbl(id serial primary key, data text);
  23 CREATE TABLE target_tbl(id serial primary key, data text);
  24
  25 SELECT pg_replication_origin_create('regress_test_decoding: regression_slot');
  26 -- ensure duplicate creations fail
  27 SELECT pg_replication_origin_create('regress_test_decoding: regression_slot');
  28
  29 --ensure deletions work (once)
  30 SELECT pg_replication_origin_create('regress_test_decoding: temp');
  31 SELECT pg_replication_origin_drop('regress_test_decoding: temp');
  32 SELECT pg_replication_origin_drop('regress_test_decoding: temp');
  33
  34 -- specifying reserved origin names is not supported
  35 SELECT pg_replication_origin_create('any');
  36 SELECT pg_replication_origin_create('none');
  37 SELECT pg_replication_origin_create('pg_replication_origin');
  38
  39 -- various failure checks for undefined slots
  40 select pg_replication_origin_advance('regress_test_decoding: temp', '0/1');
  41 select pg_replication_origin_session_setup('regress_test_decoding: temp');
  42 select pg_replication_origin_progress('regress_test_decoding: temp', true);
  43
  44 SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');
  45
  46 -- origin tx
  47 INSERT INTO origin_tbl(data) VALUES ('will be replicated and decoded and decoded again');
  48 INSERT INTO target_tbl(data)
  49 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
  50
  51 -- as is normal, the insert into target_tbl shows up
  52 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
  53
  54 INSERT INTO origin_tbl(data) VALUES ('will be replicated, but not decoded again');
  55
  56 -- mark session as replaying
  57 SELECT pg_replication_origin_session_setup('regress_test_decoding: regression_slot');
  58
  59 -- ensure we prevent duplicate setup
  60 SELECT pg_replication_origin_session_setup('regress_test_decoding: regression_slot');
  61
  62 SELECT '' FROM pg_logical_emit_message(false, 'test', 'this message will not be decoded');
  63
  64 BEGIN;
  65 -- setup transaction origin
  66 SELECT pg_replication_origin_xact_setup('0/aabbccdd', '2013-01-01 00:00');
  67 INSERT INTO target_tbl(data)
  68 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'only-local', '1');
  69 COMMIT;
  70
  71 -- check replication progress for the session is correct
  72 SELECT pg_replication_origin_session_progress(false);
  73 SELECT pg_replication_origin_session_progress(true);
  74
  75 SELECT pg_replication_origin_session_reset();
  76
  77 SELECT local_id, external_id, remote_lsn, local_lsn <> '0/0' FROM pg_replication_origin_status;
  78
  79 -- check replication progress identified by name is correct
  80 SELECT pg_replication_origin_progress('regress_test_decoding: regression_slot', false);
  81 SELECT pg_replication_origin_progress('regress_test_decoding: regression_slot', true);
  82
  83 -- ensure reset requires previously setup state
  84 SELECT pg_replication_origin_session_reset();
  85
  86 -- and magically the replayed xact will be filtered!
  87 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'only-local', '1');
  88
  89 --but new original changes still show up
  90 INSERT INTO origin_tbl(data) VALUES ('will be replicated');
  91 SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1',  'only-local', '1');
  92
  93 SELECT pg_drop_replication_slot('regression_slot');
  94 SELECT pg_replication_origin_drop('regress_test_decoding: regression_slot');
  95
  96 -- Set of transactions with no origin LSNs and commit timestamps set for
  97 -- this session.
  98 SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_no_lsn', 'test_decoding');
  99 SELECT pg_replication_origin_create('regress_test_decoding: regression_slot_no_lsn');
 100 -- mark session as replaying
 101 SELECT pg_replication_origin_session_setup('regress_test_decoding: regression_slot_no_lsn');
 102 -- Simple transactions
 103 BEGIN;
 104 INSERT INTO origin_tbl(data) VALUES ('no_lsn, commit');
 105 COMMIT;
 106 BEGIN;
 107 INSERT INTO origin_tbl(data) VALUES ('no_lsn, rollback');
 108 ROLLBACK;
 109 -- 2PC transactions
 110 BEGIN;
 111 INSERT INTO origin_tbl(data) VALUES ('no_lsn, commit prepared');
 112 PREPARE TRANSACTION 'replorigin_prepared';
 113 COMMIT PREPARED 'replorigin_prepared';
 114 BEGIN;
 115 INSERT INTO origin_tbl(data) VALUES ('no_lsn, rollback prepared');
 116 PREPARE TRANSACTION 'replorigin_prepared';
 117 ROLLBACK PREPARED 'replorigin_prepared';
 118 SELECT local_id, external_id,
 119        remote_lsn <> '0/0' AS valid_remote_lsn,
 120        local_lsn <> '0/0' AS valid_local_lsn
 121        FROM pg_replication_origin_status;
 122 SELECT data FROM pg_logical_slot_get_changes('regression_slot_no_lsn', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0');
 123 -- Clean up
 124 SELECT pg_replication_origin_session_reset();
 125 SELECT pg_drop_replication_slot('regression_slot_no_lsn');
 126 SELECT pg_replication_origin_drop('regress_test_decoding: regression_slot_no_lsn');
 127
 128 -- Test that the pgoutput correctly filters changes corresponding to the provided origin value.
 129 SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'pgoutput');
 130 CREATE PUBLICATION pub FOR TABLE target_tbl;
 131 SELECT pg_replication_origin_create('regress_test_decoding: regression_slot');
 132
 133 -- mark session as replaying
 134 SELECT pg_replication_origin_session_setup('regress_test_decoding: regression_slot');
 135
 136 INSERT INTO target_tbl(data) VALUES ('test data');
 137
 138 -- The replayed change will be filtered.
 139 SELECT count(*) = 0 FROM pg_logical_slot_peek_binary_changes('regression_slot', NULL, NULL, 'proto_version', '4', 'publication_names', 'pub', 'origin', 'none');
 140
 141 -- The replayed change will be output if the origin value is not specified.
 142 SELECT count(*) != 0 FROM pg_logical_slot_peek_binary_changes('regression_slot', NULL, NULL, 'proto_version', '4', 'publication_names', 'pub');
 143
 144 -- Clean up
 145 SELECT pg_replication_origin_session_reset();
 146 SELECT pg_drop_replication_slot('regression_slot');
 147 SELECT pg_replication_origin_drop('regress_test_decoding: regression_slot');
 148 DROP PUBLICATION pub;