CREATE EXTENSION logical_ddl;
CREATE TABLE public.replicated_table1 (id bigint primary key, c_text text);
INSERT INTO logical_ddl.settings
    VALUES (false, 'publisher1');
INSERT INTO logical_ddl.subscribe_tablelist (source, relid)
    VALUES ('publisher1', 'public.replicated_table1'::regclass);
SET session_replication_role TO 'replica';
--Table renaming test
INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"simple","command_tag":"alter table","sub_command_type":"rename table","table_name":"replicated_table1","column_name":"","new_name":"renamed_replicated_table1","column_type":""}'
    );
SELECT c.relname, c.relnamespace::regnamespace
    FROM pg_catalog.pg_class c
    WHERE
        c.relname = 'renamed_replicated_table1' AND
        c.relnamespace = 'public'::regnamespace;
          relname          | relnamespace 
---------------------------+--------------
 renamed_replicated_table1 | public
(1 row)

--Column renaming test
INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"simple","command_tag":"alter table","sub_command_type":"rename column","table_name":"renamed_replicated_table1","column_name":"c_text","new_name":"c_text_renamed","column_type":""}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_text_renamed'
    ORDER BY a.attnum;
    attname     | format_type 
----------------+-------------
 c_text_renamed | text
(1 row)

--Column adding tests
INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_char","new_name":"","column_type":"character(1)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_char'
    ORDER BY a.attnum;
 attname | format_type  
---------+--------------
 c_char  | character(1)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_char20","new_name":"","column_type":"character(20)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_char20'
    ORDER BY a.attnum;
 attname  |  format_type  
----------+---------------
 c_char20 | character(20)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_varchar","new_name":"","column_type":"character varying"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_varchar'
    ORDER BY a.attnum;
  attname  |    format_type    
-----------+-------------------
 c_varchar | character varying
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_varchar20","new_name":"","column_type":"character varying(20)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_varchar20'
    ORDER BY a.attnum;
   attname   |      format_type      
-------------+-----------------------
 c_varchar20 | character varying(20)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_text","new_name":"","column_type":"text"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_text'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_text  | text
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_smallint","new_name":"","column_type":"smallint"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_smallint'
    ORDER BY a.attnum;
  attname   | format_type 
------------+-------------
 c_smallint | smallint
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_int","new_name":"","column_type":"integer"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_int'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_int   | integer
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_bigint","new_name":"","column_type":"bigint"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_bigint'
    ORDER BY a.attnum;
 attname  | format_type 
----------+-------------
 c_bigint | bigint
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_bit","new_name":"","column_type":"bit(1)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_bit'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_bit   | bit(1)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_bit20","new_name":"","column_type":"bit(20)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_bit20'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_bit20 | bit(20)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_bitvarying","new_name":"","column_type":"bit varying"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_bitvarying'
    ORDER BY a.attnum;
   attname    | format_type 
--------------+-------------
 c_bitvarying | bit varying
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_bitvarying20","new_name":"","column_type":"bit varying(20)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_bitvarying20'
    ORDER BY a.attnum;
    attname     |   format_type   
----------------+-----------------
 c_bitvarying20 | bit varying(20)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_boolean","new_name":"","column_type":"boolean"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_boolean'
    ORDER BY a.attnum;
  attname  | format_type 
-----------+-------------
 c_boolean | boolean
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_bytea","new_name":"","column_type":"bytea"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_bytea'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_bytea | bytea
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_cidr","new_name":"","column_type":"cidr"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_cidr'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_cidr  | cidr
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_date","new_name":"","column_type":"date"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_date'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_date  | date
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_timestamp","new_name":"","column_type":"timestamp without time zone"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_timestamp'
    ORDER BY a.attnum;
   attname   |         format_type         
-------------+-----------------------------
 c_timestamp | timestamp without time zone
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_timestamp2","new_name":"","column_type":"timestamp(2) without time zone"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_timestamp2'
    ORDER BY a.attnum;
   attname    |          format_type           
--------------+--------------------------------
 c_timestamp2 | timestamp(2) without time zone
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_timestamptz","new_name":"","column_type":"timestamp with time zone"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_timestamptz'
    ORDER BY a.attnum;
    attname    |       format_type        
---------------+--------------------------
 c_timestamptz | timestamp with time zone
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_timestamptz2","new_name":"","column_type":"timestamp(2) with time zone"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_timestamptz2'
    ORDER BY a.attnum;
    attname     |         format_type         
----------------+-----------------------------
 c_timestamptz2 | timestamp(2) with time zone
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_time","new_name":"","column_type":"time without time zone"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_time'
    ORDER BY a.attnum;
 attname |      format_type       
---------+------------------------
 c_time  | time without time zone
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_time2","new_name":"","column_type":"time(2) without time zone"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_time2'
    ORDER BY a.attnum;
 attname |        format_type        
---------+---------------------------
 c_time2 | time(2) without time zone
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_timetz","new_name":"","column_type":"time with time zone"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_timetz'
    ORDER BY a.attnum;
 attname  |     format_type     
----------+---------------------
 c_timetz | time with time zone
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_timetz2","new_name":"","column_type":"time(2) with time zone"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_timetz2'
    ORDER BY a.attnum;
  attname  |      format_type       
-----------+------------------------
 c_timetz2 | time(2) with time zone
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_interval","new_name":"","column_type":"interval"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_interval'
    ORDER BY a.attnum;
  attname   | format_type 
------------+-------------
 c_interval | interval
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_interval2","new_name":"","column_type":"interval(2)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_interval2'
    ORDER BY a.attnum;
   attname   | format_type 
-------------+-------------
 c_interval2 | interval(2)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_interval_hour","new_name":"","column_type":"interval hour to second"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_interval_hour'
    ORDER BY a.attnum;
     attname     |       format_type       
-----------------+-------------------------
 c_interval_hour | interval hour to second
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_interval_hour2","new_name":"","column_type":"interval hour to second(2)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_interval_hour2'
    ORDER BY a.attnum;
     attname      |        format_type         
------------------+----------------------------
 c_interval_hour2 | interval hour to second(2)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_json","new_name":"","column_type":"json"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_json'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_json  | json
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_jsonb","new_name":"","column_type":"jsonb"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_jsonb'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_jsonb | jsonb
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_numeric","new_name":"","column_type":"numeric"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_numeric'
    ORDER BY a.attnum;
  attname  | format_type 
-----------+-------------
 c_numeric | numeric
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_numeric10","new_name":"","column_type":"numeric(10,0)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_numeric10'
    ORDER BY a.attnum;
   attname   |  format_type  
-------------+---------------
 c_numeric10 | numeric(10,0)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_numeric10_2","new_name":"","column_type":"numeric(10,2)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_numeric10_2'
    ORDER BY a.attnum;
    attname    |  format_type  
---------------+---------------
 c_numeric10_2 | numeric(10,2)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_decimal","new_name":"","column_type":"numeric"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_decimal'
    ORDER BY a.attnum;
  attname  | format_type 
-----------+-------------
 c_decimal | numeric
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_decimal10","new_name":"","column_type":"numeric(10,0)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_decimal10'
    ORDER BY a.attnum;
   attname   |  format_type  
-------------+---------------
 c_decimal10 | numeric(10,0)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_decimal10_2","new_name":"","column_type":"numeric(10,2)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_decimal10_2'
    ORDER BY a.attnum;
    attname    |  format_type  
---------------+---------------
 c_decimal10_2 | numeric(10,2)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_real","new_name":"","column_type":"real"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_real'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_real  | real
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_double_precision","new_name":"","column_type":"double precision"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_double_precision'
    ORDER BY a.attnum;
      attname       |   format_type    
--------------------+------------------
 c_double_precision | double precision
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_money","new_name":"","column_type":"money"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_money'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_money | money
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_point","new_name":"","column_type":"point"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_point'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_point | point
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_line","new_name":"","column_type":"line"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_line'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_line  | line
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_lseg","new_name":"","column_type":"lseg"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_lseg'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_lseg  | lseg
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_box","new_name":"","column_type":"box"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_box'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_box   | box
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_path","new_name":"","column_type":"path"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_path'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_path  | path
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_polygon","new_name":"","column_type":"polygon"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_polygon'
    ORDER BY a.attnum;
  attname  | format_type 
-----------+-------------
 c_polygon | polygon
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_circle","new_name":"","column_type":"circle"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_circle'
    ORDER BY a.attnum;
 attname  | format_type 
----------+-------------
 c_circle | circle
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_inet","new_name":"","column_type":"inet"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_inet'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_inet  | inet
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_macaddr","new_name":"","column_type":"macaddr"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_macaddr'
    ORDER BY a.attnum;
  attname  | format_type 
-----------+-------------
 c_macaddr | macaddr
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_macaddr8","new_name":"","column_type":"macaddr8"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_macaddr8'
    ORDER BY a.attnum;
  attname   | format_type 
------------+-------------
 c_macaddr8 | macaddr8
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_tsvector","new_name":"","column_type":"tsvector"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_tsvector'
    ORDER BY a.attnum;
  attname   | format_type 
------------+-------------
 c_tsvector | tsvector
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_tsquery","new_name":"","column_type":"tsquery"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_tsquery'
    ORDER BY a.attnum;
  attname  | format_type 
-----------+-------------
 c_tsquery | tsquery
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_uuid","new_name":"","column_type":"uuid"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_uuid'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_uuid  | uuid
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_xml","new_name":"","column_type":"xml"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_xml'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
 c_xml   | xml
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_int4range","new_name":"","column_type":"int4range"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_int4range'
    ORDER BY a.attnum;
   attname   | format_type 
-------------+-------------
 c_int4range | int4range
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_int8range","new_name":"","column_type":"int8range"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_int8range'
    ORDER BY a.attnum;
   attname   | format_type 
-------------+-------------
 c_int8range | int8range
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_numrange","new_name":"","column_type":"numrange"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_numrange'
    ORDER BY a.attnum;
  attname   | format_type 
------------+-------------
 c_numrange | numrange
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_tsrange","new_name":"","column_type":"tsrange"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_tsrange'
    ORDER BY a.attnum;
  attname  | format_type 
-----------+-------------
 c_tsrange | tsrange
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_tstzrange","new_name":"","column_type":"tstzrange"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_tstzrange'
    ORDER BY a.attnum;
   attname   | format_type 
-------------+-------------
 c_tstzrange | tstzrange
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_daterange","new_name":"","column_type":"daterange"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_daterange'
    ORDER BY a.attnum;
   attname   | format_type 
-------------+-------------
 c_daterange | daterange
(1 row)

--Column data type changing tests
INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"alter column type","table_name":"renamed_replicated_table1","column_name":"c_char","new_name":"","column_type":"character varying(20)"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_char'
    ORDER BY a.attnum;
 attname |      format_type      
---------+-----------------------
 c_char  | character varying(20)
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"alter column type","table_name":"renamed_replicated_table1","column_name":"c_varchar","new_name":"","column_type":"text"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_varchar'
    ORDER BY a.attnum;
  attname  | format_type 
-----------+-------------
 c_varchar | text
(1 row)

--Column dropping tests
INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"drop column","table_name":"renamed_replicated_table1","column_name":"c_int8range","new_name":"","column_type":""}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = true AND
        a.attname = 'c_int8range'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
(0 rows)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"drop column","table_name":"renamed_replicated_table1","column_name":"id","new_name":"","column_type":""}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = true AND
        a.attname = 'id'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
(0 rows)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"drop column","table_name":"renamed_replicated_table1","column_name":"c_text_renamed","new_name":"","column_type":""}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = true AND
        a.attname = 'c_text_renamed'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
(0 rows)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"drop column","table_name":"renamed_replicated_table1","column_name":"c_macaddr","new_name":"","column_type":""}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = true AND
        a.attname = 'c_macaddr'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
(0 rows)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"drop column","table_name":"renamed_replicated_table1","column_name":"c_text","new_name":"","column_type":""}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = true AND
        a.attname = 'c_text'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
(0 rows)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"drop column","table_name":"renamed_replicated_table1","column_name":"c_bigint","new_name":"","column_type":""}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = true AND
        a.attname = 'c_bigint'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
(0 rows)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"drop column","table_name":"renamed_replicated_table1","column_name":"c_jsonb","new_name":"","column_type":""}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = true AND
        a.attname = 'c_jsonb'
    ORDER BY a.attnum;
 attname | format_type 
---------+-------------
(0 rows)

--Multiple subcommand in one alter table command
INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_msc_text","new_name":"","column_type":"text"}'
    ),
    (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_msc_int","new_name":"","column_type":"integer"}'
    ),
    (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_msc_bigint","new_name":"","column_type":"bigint"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname LIKE 'c_msc%'
    ORDER BY a.attnum;
   attname    | format_type 
--------------+-------------
 c_msc_text   | text
 c_msc_int    | integer
 c_msc_bigint | bigint
(3 rows)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"drop column","table_name":"renamed_replicated_table1","column_name":"c_msc_int","new_name":"","column_type":""}'
    ),
    (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_msc_timestamp","new_name":"","column_type":"timestamp without time zone"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        (a.attisdropped, a.attname) IN (
            (true, 'c_msc_int'),
            (false, 'c_msc_timestamp')
        )
    ORDER BY a.attnum;
     attname     |         format_type         
-----------------+-----------------------------
 c_msc_timestamp | timestamp without time zone
(1 row)

--Column adding tests with array data type
INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_bitvarying_arr","new_name":"","column_type":"bit varying[]"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_bitvarying_arr'
    ORDER BY a.attnum;
     attname      |  format_type  
------------------+---------------
 c_bitvarying_arr | bit varying[]
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_bigint_arr","new_name":"","column_type":"bigint[]"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_bigint_arr'
    ORDER BY a.attnum;
   attname    | format_type 
--------------+-------------
 c_bigint_arr | bigint[]
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_varchar_arr","new_name":"","column_type":"character varying[]"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_varchar_arr'
    ORDER BY a.attnum;
    attname    |     format_type     
---------------+---------------------
 c_varchar_arr | character varying[]
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_varchar80_arr","new_name":"","column_type":"character varying(80)[]"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_varchar80_arr'
    ORDER BY a.attnum;
     attname     |       format_type       
-----------------+-------------------------
 c_varchar80_arr | character varying(80)[]
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_numrange_arr","new_name":"","column_type":"numrange[]"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_numrange_arr'
    ORDER BY a.attnum;
    attname     | format_type 
----------------+-------------
 c_numrange_arr | numrange[]
(1 row)

INSERT INTO logical_ddl.shadow_table (source, schema_name, command)
    VALUES (
        'publisher1',
        'public',
        '{"command_type":"alter table","command_tag":"alter table","sub_command_type":"add column","table_name":"renamed_replicated_table1","column_name":"c_numeric10_2_arr","new_name":"","column_type":"numeric(10,2)[]"}'
    );
SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
    FROM pg_catalog.pg_attribute a
    WHERE
        a.attrelid = 'public.renamed_replicated_table1'::regclass AND
        a.attisdropped = false AND
        a.attname = 'c_numeric10_2_arr'
    ORDER BY a.attnum;
      attname      |   format_type   
-------------------+-----------------
 c_numeric10_2_arr | numeric(10,2)[]
(1 row)

--List of applied commands
SELECT source, command_string, is_failed
    FROM logical_ddl.applied_commands
    ORDER BY id;
   source   |                                            command_string                                            | is_failed 
------------+------------------------------------------------------------------------------------------------------+-----------
 publisher1 | ALTER TABLE public.replicated_table1 RENAME TO renamed_replicated_table1;                            | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 RENAME COLUMN c_text TO c_text_renamed;                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_char character(1);                         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_char20 character(20);                      | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_varchar character varying;                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_varchar20 character varying(20);           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_text text;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_smallint smallint;                         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_int integer;                               | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_bigint bigint;                             | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_bit bit(1);                                | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_bit20 bit(20);                             | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_bitvarying bit varying;                    | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_bitvarying20 bit varying(20);              | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_boolean boolean;                           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_bytea bytea;                               | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_cidr cidr;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_date date;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_timestamp timestamp without time zone;     | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_timestamp2 timestamp(2) without time zone; | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_timestamptz timestamp with time zone;      | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_timestamptz2 timestamp(2) with time zone;  | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_time time without time zone;               | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_time2 time(2) without time zone;           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_timetz time with time zone;                | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_timetz2 time(2) with time zone;            | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_interval interval;                         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_interval2 interval(2);                     | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_interval_hour interval hour to second;     | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_interval_hour2 interval hour to second(2); | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_json json;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_jsonb jsonb;                               | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_numeric numeric;                           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_numeric10 numeric(10,0);                   | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_numeric10_2 numeric(10,2);                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_decimal numeric;                           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_decimal10 numeric(10,0);                   | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_decimal10_2 numeric(10,2);                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_real real;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_double_precision double precision;         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_money money;                               | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_point point;                               | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_line line;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_lseg lseg;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_box box;                                   | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_path path;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_polygon polygon;                           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_circle circle;                             | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_inet inet;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_macaddr macaddr;                           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_macaddr8 macaddr8;                         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_tsvector tsvector;                         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_tsquery tsquery;                           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_uuid uuid;                                 | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_xml xml;                                   | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_int4range int4range;                       | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_int8range int8range;                       | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_numrange numrange;                         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_tsrange tsrange;                           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_tstzrange tstzrange;                       | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_daterange daterange;                       | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ALTER COLUMN c_char TYPE character varying(20);         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ALTER COLUMN c_varchar TYPE text;                       | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 DROP COLUMN c_int8range;                                | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 DROP COLUMN id;                                         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 DROP COLUMN c_text_renamed;                             | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 DROP COLUMN c_macaddr;                                  | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 DROP COLUMN c_text;                                     | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 DROP COLUMN c_bigint;                                   | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 DROP COLUMN c_jsonb;                                    | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_msc_text text;                             | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_msc_int integer;                           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_msc_bigint bigint;                         | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 DROP COLUMN c_msc_int;                                  | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_msc_timestamp timestamp without time zone; | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_bitvarying_arr bit varying[];              | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_bigint_arr bigint[];                       | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_varchar_arr character varying[];           | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_varchar80_arr character varying(80)[];     | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_numrange_arr numrange[];                   | f
 publisher1 | ALTER TABLE public.renamed_replicated_table1 ADD COLUMN c_numeric10_2_arr numeric(10,2)[];           | f
(81 rows)

DROP EXTENSION logical_ddl CASCADE;
DROP TABLE IF EXISTS public.renamed_replicated_table1;
DROP TABLE IF EXISTS public.replicated_table1;
NOTICE:  table "replicated_table1" does not exist, skipping