/*
 * ---OnlineExpansion regression test suits---
 * Unit tests for delete delta mechanism
 */
set enable_nodegroup_debug=on;
set expected_computing_nodegroup=group_b;
create node group group_a with (datanode1,datanode2,datanode3,datanode4);
create node group group_b with (datanode1,datanode2,datanode3,datanode4,datanode5,datanode6);
CREATE SCHEMA data_redis;
CREATE OR REPLACE FUNCTION pg_tupleid_get_blocknum(tid) RETURNS int8 AS '@abs_bindir@/../lib/postgresql/gsredistribute.so', 'pg_tupleid_get_blocknum' LANGUAGE C STABLE NOT FENCED;
CREATE OR REPLACE FUNCTION pg_tupleid_get_offset(tid) RETURNS int AS '@abs_bindir@/../lib/postgresql/gsredistribute.so', 'pg_tupleid_get_offset' LANGUAGE C STABLE NOT FENCED;

/* Verify delete_delta mechanism via DELETE operation */
create table tx (c1 int, c2 int, c3 int) distribute by hash (c1) to group group_a;
create table tx_verify(c1 int, c2 int, c3 int, xcnodeid int,dnrelid int, block_number bigint, block_offset int) distribute by hash (c1) to group group_b;
create unlogged table data_redis.pg_delete_delta_12345
(
	xcnodeid int,
	dnrelid int,
	block_number bigint,
	block_offset int
)
distribute by hash(xcnodeid,dnrelid,block_number,block_offset) to group group_a;

insert into tx select v,v,v from generate_series(1,30) as v;
insert into tx_verify select *, xc_node_id, tableoid, pg_tupleid_get_blocknum(ctid), pg_tupleid_get_offset(ctid) from tx;

/*
 * DELETE rows in tx via regular DELETE operation and meanwhile make the deleted rows
 * are populated in data_redis.pg_delete_delta_12345 and using DELETE-USING on tx_verify
 */
ALTER TABLE tx SET (append_mode=on,rel_cn_oid=12345);
delete from tx where c1 < 15;
ALTER TABLE tx SET (append_mode=off);

-- delete tx_verify via delete-using clause
delete from tx_verify as t1 using data_redis.pg_delete_delta_12345 as t2
where t1.xcnodeid = t2.xcnodeid and
	  t1.dnrelid = t2.dnrelid and
	  t1.block_number = t2.block_number and
	  t1.block_offset = t2.block_offset   
	  ;

-- verify if deleted rows are record into pg_delete_delta table
select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx
   minus all
   select c1,c2,c3 from tx_verify
);

select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx_verify
   minus all
   select c1,c2,c3 from tx
);

-- error path test expecting error
alter table tx set (append_mode=on,rel_cn_oid=12345); 
drop table data_redis.pg_delete_delta_12345;
delete from tx where c1 < 15;

drop table tx;
drop table tx_verify;
--------------------------------------------------------
/* Verify delete_delta mechanism via UPDATE operation */
create table tx(c1 int, c2 int, c3 int) distribute by hash (c1) to group group_a;
create table tx_verify(c1 int, c2 int, c3 int, xcnodeid int,dnrelid int, block_number bigint, block_offset int) distribute by hash (c1) to group group_b;
create unlogged table data_redis.pg_delete_delta_12345
(
	xcnodeid int,
	dnrelid int,
	block_number bigint,
	block_offset int
)
distribute by hash(xcnodeid,dnrelid,block_number,block_offset) to group group_a;

insert into tx select v,v,v from generate_series(1,30) as v;
insert into tx_verify select *, xc_node_id, tableoid, pg_tupleid_get_blocknum(ctid), pg_tupleid_get_offset(ctid) from tx;

ALTER TABLE tx SET (append_mode=on,rel_cn_oid=12345);
update tx set c2 = 999 where c1 > 15;
ALTER TABLE tx SET (append_mode=off);

-- update tx_verify via update-from clause
update tx_verify as t1 set c2 = 999 from data_redis.pg_delete_delta_12345 as t2
where t1.xcnodeid = t2.xcnodeid and
	  t1.dnrelid = t2.dnrelid and
	  t1.block_number = t2.block_number and
	  t1.block_offset = t2.block_offset   
	  ;

-- verify if updated rows are record into pg_delete_delta table
select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx
   minus all
   select c1,c2,c3 from tx_verify
);

select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx_verify
   minus all
   select c1,c2,c3 from tx
);

-- error path test expecting error
alter table tx set (append_mode=on,rel_cn_oid=12345); 
drop table data_redis.pg_delete_delta_12345;
update tx set c2 = 999 where c1 > 15;

drop table tx;
drop table tx_verify;
------------------------------------------
/* Verify delete join operation */
create table tx(c1 int, c2 int, c3 int) distribute by hash (c1) to group group_a;
create table rowtable(c1 int, c2 int, c3 int);
create table tx_verify(c1 int, c2 int, c3 int, xcnodeid int,dnrelid int, block_number bigint, block_offset int) distribute by hash (c1) to group group_b;
create unlogged table data_redis.pg_delete_delta_12345
(
	xcnodeid int,
	dnrelid int,
	block_number bigint,
	block_offset int
)
distribute by hash(xcnodeid,dnrelid,block_number,block_offset) to group group_a;

insert into tx select v,v,v from generate_series(1,30) as v;
insert into tx_verify select *, xc_node_id, tableoid, pg_tupleid_get_blocknum(ctid), pg_tupleid_get_offset(ctid) from tx;
insert into rowtable values(10,10,10),(20,20,20),(30,30,30);

ALTER TABLE tx SET (append_mode=on,rel_cn_oid=12345);
delete from tx as t1 using rowtable as t2 
where t1.c1 = t2.c1;
ALTER TABLE tx SET (append_mode=off);

delete from tx_verify as t1 using data_redis.pg_delete_delta_12345 as t2
where t1.xcnodeid = t2.xcnodeid and
	  t1.dnrelid = t2.dnrelid and
	  t1.block_number = t2.block_number and
	  t1.block_offset = t2.block_offset   
	  ;

-- verify if updated rows are record into pg_delete_delta table
select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx
   minus all
   select c1,c2,c3 from tx_verify
);

select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx_verify
   minus all
   select c1,c2,c3 from tx
);

drop table tx;
drop table tx_verify;
drop table data_redis.pg_delete_delta_12345;
drop table rowtable;
------------------------------------------
/* Verify update join operation */
create table tx(c1 int, c2 int, c3 int) distribute by hash (c1) to group group_a;
create table rowtable(c1 int, c2 int, c3 int);
create table tx_verify(c1 int, c2 int, c3 int, xcnodeid int,dnrelid int, block_number bigint, block_offset int) distribute by hash (c1) to group group_b;
create unlogged table data_redis.pg_delete_delta_12345
(
	xcnodeid int,
	dnrelid int,
	block_number bigint,
	block_offset int
)
distribute by hash(xcnodeid,dnrelid,block_number,block_offset) to group group_a;

insert into tx select v,v,v from generate_series(1,30) as v;
insert into tx_verify select *, xc_node_id, tableoid, pg_tupleid_get_blocknum(ctid), pg_tupleid_get_offset(ctid) from tx;
insert into rowtable values(10,10,10),(20,20,20),(30,30,30);

ALTER TABLE tx SET (append_mode=on,rel_cn_oid=12345);
update tx as t1 set c2 = 1 from rowtable as t2
where t1.c1 = t2.c1;
ALTER TABLE tx SET (append_mode=off);

update tx_verify as t1 set c2 = 1 from data_redis.pg_delete_delta_12345 as t2
where t1.xcnodeid = t2.xcnodeid and
	  t1.dnrelid = t2.dnrelid and
	  t1.block_number = t2.block_number and
	  t1.block_offset = t2.block_offset   
	  ;

-- verify if updated rows are record into pg_delete_delta table
select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx
   minus all
   select c1,c2,c3 from tx_verify
);

select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx_verify
   minus all
   select c1,c2,c3 from tx
);

drop table tx;
drop table tx_verify;
drop table data_redis.pg_delete_delta_12345;
drop table rowtable;
------------------------------------------
/* Verify replicated table delete */
create table tx(c1 int, c2 int, c3 int) distribute by replication to group group_a;
create table tx_verify(c1 int, c2 int, c3 int, xcnodeid int,dnrelid int, block_number bigint, block_offset int) distribute by replication to group group_b;
ALTER TABLE tx_verify ADD CONSTRAINT uniq_con PRIMARY KEY (xcnodeid,dnrelid,block_number,block_offset);

create unlogged table data_redis.pg_delete_delta_12345
(
	xcnodeid int,
	dnrelid int,
	block_number bigint,
	block_offset int
)
distribute by hash(xcnodeid,dnrelid,block_number,block_offset) to group group_a;


insert into tx select v,v,v from generate_series(1,30) as v;
insert into tx_verify select *, xc_node_id, tableoid, pg_tupleid_get_blocknum(ctid), pg_tupleid_get_offset(ctid) from tx;

ALTER TABLE tx SET (append_mode=on,rel_cn_oid=12345);
delete from tx where c1 < 15;
ALTER TABLE tx SET (append_mode=off);

-- delete tx_verify via delete-using clause
delete from tx_verify as t1 using data_redis.pg_delete_delta_12345 as t2
where t1.xcnodeid = t2.xcnodeid and
	  t1.dnrelid = t2.dnrelid and
	  t1.block_number = t2.block_number and
	  t1.block_offset = t2.block_offset   
	  ;

-- verify if updated rows are record into pg_delete_delta table
select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx
   minus all
   select c1,c2,c3 from tx_verify
);

select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx_verify
   minus all
   select c1,c2,c3 from tx
);

drop table tx;
drop table tx_verify;
drop table data_redis.pg_delete_delta_12345;

------------------------------------------
/* Verify replicated table update*/
create table tx(c1 int, c2 int, c3 int) distribute by replication to group group_a;
create table tx_verify(c1 int, c2 int, c3 int, xcnodeid int,dnrelid int, block_number bigint, block_offset int) distribute by replication to group group_b;
ALTER TABLE tx_verify ADD CONSTRAINT uniq_con PRIMARY KEY (xcnodeid,dnrelid,block_number,block_offset);
create unlogged table data_redis.pg_delete_delta_12345
(
	xcnodeid int,
	dnrelid int,
	block_number bigint,
	block_offset int
)
distribute by hash(xcnodeid,dnrelid,block_number,block_offset) to group group_a;

insert into tx select v,v,v from generate_series(1,30) as v;
insert into tx_verify select *, xc_node_id, tableoid, pg_tupleid_get_blocknum(ctid), pg_tupleid_get_offset(ctid) from tx;

ALTER TABLE tx SET (append_mode=on,rel_cn_oid=12345);
update tx set c2 = c2*2 where c1 > 15;
ALTER TABLE tx SET (append_mode=off);

-- delete tx_verify via delete-using clause
update tx_verify as t1 set c2 = c2*2 from data_redis.pg_delete_delta_12345 as t2
where t1.xcnodeid = t2.xcnodeid and
	  t1.dnrelid = t2.dnrelid and
	  t1.block_number = t2.block_number and
	  t1.block_offset = t2.block_offset   
	  ;

-- verify if updated rows are record into pg_delete_delta table
select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx
   minus all
   select c1,c2,c3 from tx_verify
);

select 'Check result return code: ' || count(*)
from
(
   select c1,c2,c3 from tx_verify
   minus all
   select c1,c2,c3 from tx
);

drop table tx;
drop table tx_verify;
drop table data_redis.pg_delete_delta_12345;

drop function pg_tupleid_get_blocknum;
drop function pg_tupleid_get_offset;

drop node group group_a;
drop node group group_b;
DROP SCHEMA data_redis CASCADE;
reset expected_computing_nodegroup;
reset enable_nodegroup_debug;