row_filter_ddl.sql

SELECT * FROM pglogical_regress_variables()
\gset

\c :provider_dsn

SELECT pglogical.replicate_ddl_command($$
	CREATE TABLE public.rowfilter_ddl_tbl (id int primary key);
$$);

INSERT INTO rowfilter_ddl_tbl SELECT g FROM generate_series(1, 5) g;

create or replace function funcn(id int) returns bool as $$
declare
  x integer;
begin
  if to_regclass('rowfilter_ddl_tbl') is not null then
    x := (select count(*) from rowfilter_ddl_tbl);
  end if;
  raise log 'called with version a: % count %', id, x;
  return true;
end
$$ language plpgsql volatile;

SELECT * FROM pglogical.replication_set_add_table('default', 'rowfilter_ddl_tbl', true, row_filter := $rf$ funcn(id) $rf$);

insert into rowfilter_ddl_tbl values (10);

create or replace function funcn(id int) returns bool as $$
declare
  x integer;
begin
  if to_regclass('rowfilter_ddl_tbl') is not null then
    x := (select count(*) from rowfilter_ddl_tbl);
  end if;
  raise log 'called with version b: % count %', id, x;
  return true;
end
$$ language plpgsql volatile;

SELECT * FROM rowfilter_ddl_tbl ORDER BY id limit 5;
SELECT pglogical.wait_slot_confirm_lsn(NULL, NULL);

insert into rowfilter_ddl_tbl values (11);

\c :subscriber_dsn

BEGIN;
SET LOCAL statement_timeout = '180s';
SELECT pglogical.wait_for_table_sync_complete('test_subscription', 'rowfilter_ddl_tbl');
COMMIT;

SELECT sync_kind, sync_nspname, sync_relname, sync_status FROM pglogical.local_sync_status WHERE sync_relname = 'rowfilter_ddl_tbl';

SELECT * FROM rowfilter_ddl_tbl;


\c :provider_dsn

begin;
insert into rowfilter_ddl_tbl select g+20 from generate_series(1, 10) g;
create or replace function funcn(id int) returns bool as $$
declare
  x integer;
begin
  if to_regclass('rowfilter_ddl_tbl') is not null then
    x := (select count(*) from rowfilter_ddl_tbl);
  end if;
  raise log 'called with version c: % count %', id, x;
  return true;
end
$$ language plpgsql volatile;

insert into rowfilter_ddl_tbl values (50);

-- ALTER the table, while decoding is active
alter table rowfilter_ddl_tbl alter column id type text;

create or replace function funcn(id int) returns bool as $$
declare
  x integer = null;
begin
  if to_regclass('rowfilter_ddl_tbl') is not null then
    x := (select count(*) from rowfilter_ddl_tbl);
  end if;
  raise log 'called with version d: % count %', id, x;
  return true;
end
$$ language plpgsql volatile;
commit;
insert into rowfilter_ddl_tbl values (60);

\c :subscriber_dsn

SELECT pglogical.wait_slot_confirm_lsn(NULL, NULL);
SELECT * FROM rowfilter_ddl_tbl;