# 构成Canal HA的126、127两台都执行
~/canal_113/deployer/bin/stop.sh
# 查询Zookeeper确认
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/zookeeper/bin/zkCli.sh
[zk: localhost:2181(CONNECTED) 0] ls /otter/canal/destinations/example/cluster
[]
# 停止Canal Adapter，126执行
~/canal_113/adapter/bin/stop.sh

-- 建立源数据库，126主库上执行
drop database if exists source;  
create database source;  
 
use source;  
 
-- 建立客户表    
create table customer (    
    customer_number int not null auto_increment primary key comment '客户编号，主键',    
    customer_name varchar(50) comment '客户名称',    
    customer_street_address varchar(50) comment '客户住址',    
    customer_zip_code int comment '邮编',    
    customer_city varchar(30) comment '所在城市',    
    customer_state varchar(2) comment '所在省份'    
);  
 
-- 建立产品表    
create table product (       
    product_code int not null auto_increment primary key comment '产品编码，主键',    
    product_name varchar(30) comment '产品名称',    
    product_category varchar(30) comment '产品类型'       
);  
 
-- 建立销售订单表    
create table sales_order (    
    order_number bigint not null auto_increment primary key comment '订单号，主键',    
    customer_number int comment '客户编号',    
    product_code int comment '产品编码',    
    order_date datetime comment '订单时间',    
    entry_date datetime comment '登记时间',    
    order_amount decimal(10 , 2 ) comment '销售金额',    
    foreign key (customer_number)    
        references customer (customer_number)    
        on delete cascade on update cascade,    
    foreign key (product_code)    
        references product (product_code)    
        on delete cascade on update cascade    
);

use source;  
 
-- 生成客户表测试数据    
insert into customer 
(customer_name,customer_street_address,customer_zip_code,
customer_city,customer_state) 
values      
('really large customers', '7500 louise dr.',17050, 'mechanicsburg','pa'),    
('small stores', '2500 woodland st.',17055, 'pittsburgh','pa'),    
('medium retailers','1111 ritter rd.',17055,'pittsburgh','pa'),    
('good companies','9500 scott st.',17050,'mechanicsburg','pa'),    
('wonderful shops','3333 rossmoyne rd.',17050,'mechanicsburg','pa'),    
('loyal clients','7070 ritter rd.',17055,'pittsburgh','pa'),    
('distinguished partners','9999 scott st.',17050,'mechanicsburg','pa');  
 
-- 生成产品表测试数据    
insert into product (product_name,product_category)    
values     
('hard disk drive', 'storage'),    
('floppy drive', 'storage'),    
('lcd panel', 'monitor');  
  
-- 生成100条销售订单表测试数据  
drop procedure if exists generate_sales_order_data;  
delimiter //    
create procedure generate_sales_order_data()  
begin   
    drop table if exists temp_sales_order_data;  
    create table temp_sales_order_data as select * from sales_order where 1=0;  
  
    set @start_date := unix_timestamp('2021-06-01');  
    set @end_date := unix_timestamp('2021-10-01');  
    set @i := 1;  
  
    while @i<=100 do  
        set @customer_number := floor(1 + rand() * 6);  
        set @product_code := floor(1 + rand() * 2);  
        set @order_date := from_unixtime(@start_date + rand() * (@end_date - @start_date));  
        set @amount := floor(1000 + rand() * 9000);  
  
        insert into temp_sales_order_data values (@i,@customer_number,@product_code,@order_date,@order_date,@amount);  
        set @i:=@i+1;  
    end while;  
  
    truncate table sales_order;  
    insert into sales_order   
    select null,customer_number,product_code,order_date,entry_date,order_amount from temp_sales_order_data order by order_date;  
    commit;  
  
end   
//    
delimiter ;   
  
call generate_sales_order_data();

# 用gpadmin用户连接Greenplum
psql
-- 创建资源队列
create resource queue rsq_dwtest with 
(active_statements=20,memory_limit='8000MB',priority=high,cost_overcommit=true,min_cost=0,max_cost=-1);
-- 修改dwtest用户使用的资源队列
alter role dwtest resource queue rsq_dwtest;
-- 查看用户资源队列
select rolname, rsqname from pg_roles, pg_resqueue  
 where pg_roles.rolresqueue=pg_resqueue.oid;

# 用dwtest用户连接Greenplum
psql -U dwtest -h mdw -d dw
-- 创建rds模式
create schema rds;
-- 创建tds模式
create schema tds;
-- 查看模式
dn
-- 修改数据库的模式查找路径
alter database dw set search_path to rds, tds, public, pg_catalog, tpcc_test; 
-- 重新连接dw数据库
c dw
-- 显示模式查找路径
show search_path;

-- 设置模式查找路径
set search_path to rds;
 
-- 建立客户原始数据表
create table customer 
( 
  customer_number int primary key,  
  customer_name varchar(30),  
  customer_street_address varchar(30),  
  customer_zip_code int,  
  customer_city varchar(30),  
  customer_state varchar(2)
); 
 
comment on table customer is '客户原始数据表';
comment on column customer.customer_number is '客户编号';
comment on column customer.customer_name is '客户姓名';
comment on column customer.customer_street_address is '客户地址';
comment on column customer.customer_zip_code is '客户邮编';
comment on column customer.customer_city is '客户所在城市';
comment on column customer.customer_state is '客户所在省份';
 
-- 建立产品原始数据表
create table product 
( 
  product_code int primary key, 
  product_name varchar(30),  
  product_category varchar(30)
); 
 
comment on table product is '产品原始数据表';
comment on column product.product_code is '产品编码';
comment on column product.product_name is '产品名称';
comment on column product.product_category is '产品类型';
 
-- 建立销售订单原始数据表
create table sales_order
( 
  order_number bigint,    
  customer_number int,    
  product_code int,    
  order_date timestamp,    
  entry_date timestamp,    
  order_amount decimal(10 , 2 ),
  primary key (order_number, entry_date)
) distributed by (order_number)
partition by range (entry_date)  
( start (date '2021-06-01') inclusive    
   end (date '2022-04-01') exclusive    
   every (interval '1 month') ); 
 
comment on table sales_order is '销售订单原始数据表';
comment on column sales_order.order_number is '订单号';
comment on column sales_order.customer_number is '客户编号';
comment on column sales_order.product_code is '产品编码';
comment on column sales_order.order_date is '订单时间';
comment on column sales_order.entry_date is '登记时间';
comment on column sales_order.order_amount is '销售金额';

ERROR:  PRIMARY KEY constraint must contain all columns in the partition key
HINT:  Include column "entry_date" in the PRIMARY KEY constraint or create a part-wise UNIQUE index after creating the table instead.

HINT:  When there is both a PRIMARY KEY and a DISTRIBUTED BY clause, the DISTRIBUTED BY clause must be a subset of the PRIMARY KEY.

-- 设置模式查找路径
set search_path to tds;
 
-- 建立客户维度表    
create table customer_dim (    
    customer_sk bigserial, 
    customer_number int,    
    customer_name varchar(50),    
    customer_street_address varchar(50),    
    customer_zip_code int,    
    customer_city varchar(30),    
    customer_state varchar(2),  
    version int,  
    effective_dt timestamp,
    expiry_dt timestamp,
    primary key (customer_sk, customer_number)    
) distributed by (customer_number);

-- 建立产品维度表    
create table product_dim (    
    product_sk bigserial, 
    product_code int,    
    product_name varchar(30),    
    product_category varchar(30),
    version int,  
    effective_dt timestamp,
    expiry_dt timestamp,
    primary key (product_sk, product_code)    
) distributed by (product_code);  
 
-- 建立订单维度表    
create table order_dim (    
    order_sk bigserial,    
    order_number bigint, 
    version int,  
    effective_dt timestamp,
    expiry_dt timestamp,
    primary key (order_sk, order_number)
) distributed by (order_number);
 
-- 建立日期维度表  
create table date_dim (      
    date_sk serial primary key,    
    date date,    
    month smallint,    
    month_name varchar(9),    
    quarter smallint,    
    year smallint );   
  
-- 建立销售订单事实表    
create table sales_order_fact (    
    order_sk bigint,    
    customer_sk bigint,    
    product_sk bigint, 
    order_date_sk int,    
    year_month int,    
    order_amount decimal(10 , 2 ),
    primary key (order_sk, customer_sk, product_sk, order_date_sk, year_month)
) distributed by (order_sk)
partition by range (year_month)  
( partition p202106 start (202106) inclusive ,
  partition p202107 start (202107) inclusive ,
  partition p202108 start (202108) inclusive ,
  partition p202109 start (202109) inclusive ,
  partition p202110 start (202110) inclusive ,
  partition p202111 start (202111) inclusive ,
  partition p202112 start (202112) inclusive ,  
  partition p202201 start (202201) inclusive ,  
  partition p202202 start (202202) inclusive ,  
  partition p202203 start (202203) inclusive   
                     end (202204) exclusive );  

-- 生成日期维度表数据的函数  
create or replace function fn_populate_date (start_dt date, end_dt date)    
returns void as $$    
declare    
    v_date date:= start_dt;   
    v_datediff int:= end_dt - start_dt;  
begin    
    for i in 0 .. v_datediff loop      
        insert into date_dim(date, month, month_name, quarter, year)    
        values(v_date, extract(month from v_date), to_char(v_date,'mon'), extract(quarter from v_date), extract(year from v_date));   
        v_date := v_date + 1;  
    end loop;  
    analyze date_dim;  
end; $$    
language plpgsql;

-- 执行函数生成日期维度数据
select fn_populate_date(date '2020-01-01', date '2022-12-31');  

-- 查询生成的日期
select min(date_sk) min_sk, min(date) min_date, max(date_sk) max_sk, max(date) max_date, count(*) c 
  from date_dim;

dw=> select last_value, start_value from date_dim_date_sk_seq;
 last_value | start_value 
------------+-------------
       1096 |           1
(1 row)

-- 127从库停复制
stop slave;
# 从从库导出数据
cd ~
mkdir -p source_bak
mysqldump -u root -p123456 -S /data/mysql.sock -t -T ~/source_bak source customer product sales_order --fields-terminated-by='|' --single-transaction
# 将导出数据文件拷贝到Greenplum的master主机
scp ~/source_bak/*.txt gpadmin@114.112.77.198:/data/source_bak/

# 用gpadmin用户连接数据库
psql -d dw

-- 设置搜索路径为rds
set search_path to rds;

-- 装载前清空表，实现幂等操作
truncate table customer;    
truncate table product;    
truncate table sales_order;    

-- 将外部数据装载到原始数据表
copy rds.customer from '/data/source_bak/customer.txt' with delimiter '|';
copy rds.product from '/data/source_bak/product.txt' with delimiter '|';
copy rds.sales_order from '/data/source_bak/sales_order.txt' with delimiter '|';

-- 装载数据后，执行查询前，先分析表以提高查询性能
analyze rds.customer;  
analyze rds.product;  
analyze rds.sales_order; 

# dwtest用户执行
psql -U dwtest -h mdw -d dw

-- 设置搜索路径
set search_path to tds, rds;

-- 装载前清空表，实现幂等操作
truncate table customer_dim;    
truncate table product_dim;    
truncate table order_dim;    

-- 序列初始化
alter sequence customer_dim_customer_sk_seq restart with 1;
alter sequence product_dim_product_sk_seq restart with 1;
alter sequence order_dim_order_sk_seq restart with 1;

-- 装载customer_dim维度表
insert into customer_dim (customer_number,customer_name,customer_street_address,customer_zip_code,customer_city,customer_state,version,effective_dt,expiry_dt) 
select customer_number,customer_name,customer_street_address,customer_zip_code,customer_city,customer_state,1,'2021-06-01','2200-01-01'
  from customer;

-- 装载product_dim维度表
insert into product_dim (product_code,product_name,product_category,version,effective_dt,expiry_dt) 
select product_code,product_name,product_category,1,'2021-06-01','2200-01-01'
  from product;

-- 装载order_dim维度表
insert into order_dim (order_number,version,effective_dt,expiry_dt)
select order_number,1,'2021-06-01','2200-01-01'
  from sales_order;

-- 装载sales_order_fact事实表
insert into sales_order_fact(order_sk,customer_sk,product_sk,order_date_sk,year_month,order_amount) 
select order_sk, customer_sk, product_sk, date_sk, to_char(order_date, 'YYYYMM')::int, order_amount
  from rds.sales_order a, order_dim b, customer_dim c, product_dim d, date_dim e
 where a.order_number = b.order_number
   and a.customer_number = c.customer_number
   and a.product_code = d.product_code
   and date(a.order_date) = e.date;

-- 分析tds模式的表  
analyze customer_dim;  
analyze product_dim;  
analyze order_dim;  
analyze sales_order_fact;

select order_number,customer_name,product_name,date,
order_amount amount  
  from sales_order_fact a, customer_dim b, product_dim c, 
order_dim d, date_dim e  
 where a.customer_sk = b.customer_sk  
   and a.product_sk = c.product_sk  
   and a.order_sk = d.order_sk  
   and a.order_date_sk = e.date_sk  
 order by order_number;  

dw=> select order_sk, order_number from order_dim order by order_number limit 5;
 order_sk | order_number 
----------+--------------
       64 |            1
       91 |            2
       90 |            3
       89 |            4
       83 |            5
(5 rows)

do $$declare
  r_mycur record;   
begin
  --在读入游标时最好先对值进行排序，保证循环调用的顺序
  for r_mycur in select order_number from sales_order order by order_number
  loop 
      --在游标内循环执行插入
      insert into order_dim (order_number,version,effective_dt,expiry_dt) 
      values (r_mycur.order_number,1,'2021-06-01','2200-01-01');
  end loop;
end$$;

customer.yml
product.yml
sales_order.yml

dataSourceKey: defaultDS
destination: example
groupId: g1
outerAdapterKey: Greenplum
concurrent: true
dbMapping:
  database: source
  table: customer
  targetTable: rds.customer
  targetPk:
    customer_number: customer_number
#  mapAll: true
  targetColumns:
    customer_number: customer_number
    customer_name: customer_name
    customer_street_address: customer_street_address
    customer_zip_code: customer_zip_code
    customer_city: customer_city
    customer_state: customer_state
  commitBatch: 30000 # 批量提交的大小

CREATE [OR REPLACE] RULE name AS ON event
  TO table_name [WHERE condition] 
  DO [ALSO | INSTEAD] { NOTHING | command | (command; command 
  ...) }

create rule r_delete_customer as on delete to customer do also 
(update customer_dim set expiry_dt=now() where customer_number=old.customer_number and expiry_dt='2200-01-01';);

create rule r_insert_customer as on insert to customer do also 
(insert into customer_dim (customer_number,customer_name,customer_street_address,customer_zip_code,customer_city,customer_state,version,effective_dt,expiry_dt) 
 values (new.customer_number,new.customer_name,new.customer_street_address,new.customer_zip_code,new.customer_city,new.customer_state,1,now(),'2200-01-01'););

create rule r_update_customer as on update to customer do also 
(update customer_dim set expiry_dt=now() 
  where customer_number=new.customer_number and expiry_dt='2200-01-01' 
    and customer_street_address <> new.customer_street_address;

insert into customer_dim (customer_number,customer_name,customer_street_address,customer_zip_code,customer_city,customer_state,version,effective_dt,expiry_dt) 
select new.customer_number,new.customer_name,new.customer_street_address,new.customer_zip_code,new.customer_city,new.customer_state,version + 1,expiry_dt,'2200-01-01'
  from customer_dim 
 where customer_number=new.customer_number 
   and customer_street_address <> new.customer_street_address 
   and version=(select max(version) 
                  from customer_dim 
                 where customer_number=new.customer_number);

update customer_dim set customer_name=new.customer_name 
 where customer_number=new.customer_number and customer_name<>new.customer_name);

create rule r_delete_product as on delete to product do also 
(update product_dim set expiry_dt=now() where product_code=old.product_code and expiry_dt='2200-01-01';);

create rule r_insert_product as on insert to product do also 
(insert into product_dim (product_code,product_name,product_category,version,effective_dt,expiry_dt) 
 values (new.product_code,new.product_name,new.product_category,1,now(),'2200-01-01'););

create rule r_update_product as on update to product do also 
(update product_dim set expiry_dt=now() 
  where product_code=new.product_code 
    and expiry_dt='2200-01-01' 
    and (product_name <> new.product_name or product_category <> new.product_category);

insert into product_dim (product_code,product_name,product_category,version,effective_dt,expiry_dt) 
select new.product_code,new.product_name,new.product_category,version + 1,expiry_dt,'2200-01-01'
  from product_dim 
 where product_code=new.product_code 
   and (product_name <> new.product_name or product_category <> new.product_category)
   and version=(select max(version) 
                  from product_dim 
                 where product_code=new.product_code));

create rule r_insert_sales_order as on insert to sales_order do also 
(insert into order_dim (order_sk, order_number,version,effective_dt,expiry_dt) 
 values (new.order_number, new.order_number,1,'2021-06-01','2200-01-01');
 
 insert into sales_order_fact(order_sk,customer_sk,product_sk,order_date_sk,year_month,order_amount) 
 select e.order_sk, customer_sk, product_sk, date_sk, to_char(order_date, 'YYYYMM')::int, order_amount
   from rds.sales_order a, customer_dim b, product_dim c, date_dim d, order_dim e
  where a.order_number = new.order_number and e.order_number = new.order_number
    and a.customer_number = b.customer_number and b.expiry_dt = '2200-01-01'
    and a.product_code = c.product_code and c.expiry_dt = '2200-01-01'
    and date(a.order_date) = d.date
 );

export PATH=$PATH:/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin/;
kafka-topics.sh --delete --topic example --bootstrap-server 172.16.1.124:9092
kafka-topics.sh --create --topic example --bootstrap-server 172.16.1.124:9092 --partitions 3 --replication-factor 3

# 在Zookeeper中删除当前数据同步位点
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/zookeeper/bin/zkCli.sh
[zk: localhost:2181(CONNECTED) 0] delete /otter/canal/destinations/example/1001/cursor

# 启动Canal Server，在构成Canal HA的126、127两台顺序执行
~/canal_113/deployer/bin/startup.sh

# 查询Zookeeper确认起始同步位点
/opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/zookeeper/bin/zkCli.sh
[zk: localhost:2181(CONNECTED) 0] get /otter/canal/destinations/example/1001/cursor
{"@type":"com.alibaba.otter.canal.protocol.position.LogPosition","identity":{"slaveId":-1,"sourceAddress":{"address":"node3","port":3306}},"postion":{"gtid":"","included":false,"journalName":"mysql-bin.000001","position":72569,"serverId":126,"timestamp":1640579962000}}

# 在126执行
~/canal_113/adapter/bin/startup.sh

-- 127从库开启复制
start slave;

use source;
  
/*** 客户数据的改变如下：  
客户6的街道号改为7777 ritter rd。（原来是7070 ritter rd），然后再改回原值。
客户7的姓名改为distinguished agencies。（原来是distinguished partners）  
新增第八个客户。 
***/  
update customer set customer_street_address = '7777 ritter rd.' where customer_number = 6 ;  
update customer set customer_street_address = '7070 ritter rd.' where customer_number = 6 ;
update customer set customer_name = 'distinguished agencies' where customer_number = 7 ;  
insert into customer (customer_name, customer_street_address, customer_zip_code, customer_city, customer_state)  
values ('subsidiaries', '10000 wetline blvd.', 17055, 'pittsburgh', 'pa') ;
  
/*** 产品数据的改变如下：  
产品3的名称改为flat panel。（原来是lcd panel）  
新增第四个产品。  
***/  
update product set product_name = 'flat panel' where product_code = 3 ;  
insert into product (product_name, product_category)  
values ('keyboard', 'peripheral') ;   

/*** 新增订单日期为2021年12月23日的16条订单。 ***/
set sql_log_bin = 0;

drop table if exists temp_sales_order_data;
create table temp_sales_order_data as select * from sales_order where 1=0;

set @start_date := unix_timestamp('2021-12-23');
set @end_date := unix_timestamp('2021-12-24');

set @customer_number := floor(1 + rand() * 8);
set @product_code := floor(1 + rand() * 4);
set @order_date := from_unixtime(@start_date + rand() * (@end_date - @start_date));
set @amount := floor(1000 + rand() * 9000);
insert into temp_sales_order_data
values (1,@customer_number,@product_code,@order_date,@order_date,@amount);

... 共插入16条数据 ...

set sql_log_bin = 1;  
insert into sales_order  
select null,customer_number,product_code,order_date,entry_date,order_amount from temp_sales_order_data order by order_date;    
  
commit ; 

2021-12-24 09:23:26.177 [pool-6-thread-1] ERROR com.alibaba.otter.canal.kafka.CanalKafkaProducer - null
java.lang.NullPointerException: null
        at com.alibaba.otter.canal.common.MQMessageUtils.messagePartition(MQMessageUtils.java:441) ~[canal.server-1.1.3.jar:na]
        at com.alibaba.otter.canal.kafka.CanalKafkaProducer.send(CanalKafkaProducer.java:174) ~[canal.server-1.1.3.jar:na]
        at com.alibaba.otter.canal.kafka.CanalKafkaProducer.send(CanalKafkaProducer.java:124) ~[canal.server-1.1.3.jar:na]
        at com.alibaba.otter.canal.server.CanalMQStarter.worker(CanalMQStarter.java:182) [canal.server-1.1.3.jar:na]
        at com.alibaba.otter.canal.server.CanalMQStarter.access$500(CanalMQStarter.java:22) [canal.server-1.1.3.jar:na]
        at com.alibaba.otter.canal.server.CanalMQStarter$CanalMQRunnable.run(CanalMQStarter.java:224) [canal.server-1.1.3.jar:na]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_232]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_232]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_232]

dw=> select customer_sk, customer_number, customer_name, customer_street_address,version,effective_dt,expiry_dt 
dw->   from customer_dim 
dw->  order by customer_number, version;
 customer_sk | customer_number |     customer_name      | customer_street_address | version |        effective_dt        |         expiry_dt          
-------------+-----------------+------------------------+-------------------------+---------+----------------------------+----------------------------
           3 |               1 | really large customers | 7500 louise dr.         |       1 | 2021-06-01 00:00:00        | 2200-01-01 00:00:00
           7 |               2 | small stores           | 2500 woodland st.       |       1 | 2021-06-01 00:00:00        | 2200-01-01 00:00:00
           6 |               3 | medium retailers       | 1111 ritter rd.         |       1 | 2021-06-01 00:00:00        | 2200-01-01 00:00:00
           1 |               4 | good companies         | 9500 scott st.          |       1 | 2021-06-01 00:00:00        | 2200-01-01 00:00:00
           5 |               5 | wonderful shops        | 3333 rossmoyne rd.      |       1 | 2021-06-01 00:00:00        | 2200-01-01 00:00:00
           2 |               6 | loyal clients          | 7070 ritter rd.         |       1 | 2021-06-01 00:00:00        | 2021-12-22 13:14:54.194449
           9 |               6 | loyal clients          | 7777 ritter rd.         |       2 | 2021-12-22 13:14:54.194449 | 2021-12-22 13:14:54.194449
          10 |               6 | loyal clients          | 7070 ritter rd.         |       3 | 2021-12-22 13:14:54.194449 | 2200-01-01 00:00:00
           4 |               7 | distinguished agencies | 9999 scott st.          |       1 | 2021-06-01 00:00:00        | 2200-01-01 00:00:00
           8 |               8 | subsidiaries           | 10000 wetline blvd.     |       1 | 2021-12-22 13:14:54.086822 | 2200-01-01 00:00:00
(10 rows)

dw=> select * from product_dim order by product_code, version;
 product_sk | product_code |  product_name   | product_category | version |        effective_dt        |         expiry_dt          
------------+--------------+-----------------+------------------+---------+----------------------------+----------------------------
          1 |            1 | hard disk drive | storage          |       1 | 2021-06-01 00:00:00        | 2200-01-01 00:00:00
          3 |            2 | floppy drive    | storage          |       1 | 2021-06-01 00:00:00        | 2200-01-01 00:00:00
          2 |            3 | lcd panel       | monitor          |       1 | 2021-06-01 00:00:00        | 2021-12-22 13:15:04.050588
          5 |            3 | flat panel      | monitor          |       2 | 2021-12-22 13:15:04.050588 | 2200-01-01 00:00:00
          4 |            4 | keyboard        | peripheral       |       1 | 2021-12-22 13:15:03.977177 | 2200-01-01 00:00:00
(5 rows)

dw=> select * from order_dim order by order_number;
 order_sk | order_number | version |    effective_dt     |      expiry_dt      
----------+--------------+---------+---------------------+---------------------
       64 |            1 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
       91 |            2 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
       90 |            3 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
       89 |            4 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
       83 |            5 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
...
      111 |          111 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
      112 |          112 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
      113 |          113 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
      114 |          114 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
      115 |          115 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
      116 |          116 |       1 | 2021-06-01 00:00:00 | 2200-01-01 00:00:00
(116 rows)

dw=> select a.order_sk, order_number,customer_name,product_name,date,order_amount
dw->   from sales_order_fact a, customer_dim b, product_dim c, order_dim d, date_dim e
dw->  where a.customer_sk = b.customer_sk
dw->    and a.product_sk = c.product_sk
dw->    and a.order_sk = d.order_sk
dw->    and a.order_date_sk = e.date_sk
dw->  order by order_number;
 order_sk | order_number |     customer_name      |  product_name   |    date    | order_amount 
----------+--------------+------------------------+-----------------+------------+--------------
...
      101 |          101 | distinguished agencies | keyboard        | 2021-12-23 |      5814.00
      102 |          102 | subsidiaries           | keyboard        | 2021-12-23 |      4362.00
      103 |          103 | distinguished agencies | hard disk drive | 2021-12-23 |      3214.00
      104 |          104 | subsidiaries           | keyboard        | 2021-12-23 |      6034.00
      105 |          105 | wonderful shops        | floppy drive    | 2021-12-23 |      8323.00
      106 |          106 | subsidiaries           | keyboard        | 2021-12-23 |      9223.00
      107 |          107 | really large customers | floppy drive    | 2021-12-23 |      5435.00
      108 |          108 | loyal clients          | floppy drive    | 2021-12-23 |      3094.00
      109 |          109 | really large customers | hard disk drive | 2021-12-23 |      8271.00
      110 |          110 | subsidiaries           | hard disk drive | 2021-12-23 |      3463.00
      111 |          111 | loyal clients          | keyboard        | 2021-12-23 |      2022.00
      112 |          112 | medium retailers       | flat panel      | 2021-12-23 |      1125.00
      113 |          113 | good companies         | hard disk drive | 2021-12-23 |      2552.00
      114 |          114 | small stores           | hard disk drive | 2021-12-23 |      5222.00
      115 |          115 | good companies         | hard disk drive | 2021-12-23 |      7801.00
      116 |          116 | really large customers | hard disk drive | 2021-12-23 |      3525.00
(116 rows)

-- 创建动态滚动分区的函数  
create or replace function tds.fn_rolling_partition(p_year_month_start date) returns int as $body$    
declare    
    v_min_partitiontablename name;  
    v_year_month_end date := p_year_month_start + interval '1 month';  
    v_year_month_start_int int := extract(year from p_year_month_start) * 100 + extract(month from p_year_month_start);  
    v_year_month_end_int int := extract(year from v_year_month_end) * 100  + extract(month from v_year_month_end);  
    sqlstring varchar(1000);      
begin   
    -- 处理rds.sales_order  
    -- 转储最早一个月的数据，    
    select partitiontablename into v_min_partitiontablename  
      from pg_partitions    
     where tablename='sales_order' and partitionrank = 1;  
   
    sqlstring = 'copy (select * from ' || v_min_partitiontablename || ') to ''/home/gpadmin/sales_order_' || cast(v_year_month_start_int as varchar) || '.txt'' with delimiter ''|'';';    
    execute sqlstring;    
    -- raise notice '%', sqlstring;    
    
    -- 删除最早月份对应的分区    
    sqlstring := 'alter table sales_order drop partition for (rank(1));';    
    execute sqlstring;    
        
    -- 增加下一个月份的新分区    
    sqlstring := 'alter table sales_order add partition start (date '''|| p_year_month_start ||''') inclusive end (date '''||v_year_month_end  ||''') exclusive;';    
    execute sqlstring;    
    -- raise notice '%', sqlstring;      

       
    -- 处理tds.sales_order_fact       
    -- 转储最早一个月的数据，    
    select partitiontablename into v_min_partitiontablename  
      from pg_partitions    
     where tablename='sales_order_fact' and partitionrank = 1;  
   
    sqlstring = 'copy (select * from ' || v_min_partitiontablename || ') to ''/home/gpadmin/sales_order_fact_' || cast(v_year_month_start_int as varchar) || '.txt'' with delimiter ''|'';';    
    execute sqlstring;    
    -- raise notice '%', sqlstring;    
    
    -- 删除最早月份对应的分区    
    sqlstring := 'alter table sales_order_fact drop partition for (rank(1));';    
    execute sqlstring;    
        
    -- 增加下一个月份的新分区    
    sqlstring := 'alter table sales_order_fact add partition start ('||cast(v_year_month_start_int as varchar)||') inclusive end ('||cast(v_year_month_end_int as varchar)||') exclusive;';    
    execute sqlstring;    
    -- raise notice '%', sqlstring;   

    -- 正常返回1    
    return 1;    
        
-- 异常返回0    
exception when others then     
    raise exception '%: %', sqlstate, sqlerrm;      
    return 0;    
end    
$body$ language plpgsql;

0 2 1 * * psql -d dw -c "set search_path=rds,tds; select fn_rolling_partition(date(date_trunc('month',current_date) + interval '1 month'));" > rolling_partition.log 2>&1