create table tb1(id int,c1 int);
select create_distributed_table('tb1','id');

postgres=# explain select count(distinct c1) from tb1;
                                        QUERY PLAN                                        
------------------------------------------------------------------------------------------
 Aggregate  (cost=250.00..250.01 rows=1 width=8)
   ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=100000 width=4)
         Task Count: 32
         Tasks Shown: One of 32
         ->  Task
               Node: host=127.0.0.1 port=9001 dbname=postgres
               ->  HashAggregate  (cost=38.25..40.25 rows=200 width=4)
                     Group Key: c1
                     ->  Seq Scan on tb1_102339 tb1  (cost=0.00..32.60 rows=2260 width=4)
(9 rows)

create table tb1(id int, c1 int);
select create_distributed_table('tb1','id');
create table tb2(id int primary key, c1 int);
select create_distributed_table('tb2','id');

postgres=# INSERT INTO tb2
  SELECT * from tb1
  ON CONFLICT(id) DO UPDATE SET c1 = EXCLUDED.c1;
INSERT 0 1

create table tbgenstore(id int, c1 int GENERATE ALWAYS AS (id+1)STORED);
select create_distributed_table('tbgenstore','id');

-- an example function which updates a hypothetical
-- event_responses table which itself is distributed by event_id
CREATE OR REPLACE FUNCTION
  register_for_event(p_event_id int, p_user_id int)
RETURNS void LANGUAGE plpgsql AS $fn$
BEGIN
  INSERT INTO event_responses VALUES ($1, $2, 'yes')
  ON CONFLICT (event_id, user_id)
  DO UPDATE SET response = EXCLUDED.response;
END;
$fn$;

-- distribute the function to workers, using the p_event_id argument
-- to determine which shard each invocation affects, and explicitly
-- colocating with event_responses which the function updates
SELECT create_distributed_function(
  'register_for_event(int, int)', 'p_event_id',
  colocate_with := 'event_responses'
);

static const char *const AggregateNames[] = {
	"invalid", "avg", "min", "max",
	"sum", "count", "array_agg",
	"jsonb_agg", "jsonb_object_agg",
	"json_agg", "json_object_agg",
	"bit_and", "bit_or", "bool_and", "bool_or", "every",
	"hll_add_agg", "hll_union_agg",
	"topn_add_agg", "topn_union_agg",
	"any_value"
};

create function sum2_sfunc(state int, x int)
returns int immutable language plpgsql as $$
begin return state + x;
end;
$$;
create function sum2_finalfunc(state int)
returns int immutable language plpgsql as $$
begin return state * 2;
end;
$$;
create aggregate sum2 (int) (
    sfunc = sum2_sfunc,
    stype = int,
    finalfunc = sum2_finalfunc,
    combinefunc = sum2_sfunc,
    initcond = '0'
);

select create_distributed_function('sum2(int)');

postgres=# explain select sum2(c1) from tb1;
                                        QUERY PLAN                                        
------------------------------------------------------------------------------------------
 Aggregate  (cost=250.00..250.01 rows=1 width=4)
   ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=100000 width=32)
         Task Count: 32
         Tasks Shown: One of 32
         ->  Task
               Node: host=127.0.0.1 port=9001 dbname=postgres
               ->  Aggregate  (cost=38.25..38.26 rows=1 width=32)
                     ->  Seq Scan on tb1_102339 tb1  (cost=0.00..32.60 rows=2260 width=4)
(8 rows)

set citus.propagate_set_commands TO local;

postgres=# begin;
BEGIN
postgres=*# set local enable_hashagg to off;
SET
postgres=*# SELECT current_setting('enable_hashagg') FROM tb1 WHERE id = 3;
 current_setting 
-----------------
 off
(1 row)

SELECT master_add_node('127.0.0.1', 9001, groupid => 0);

GRANT USAGE ON SCHEMA dist_schema TO role1;

ALTER TABLE ... SET SCHEMA

create index tb1_idx_id on tb1(id) include (c1);

create index CONCURRENTLY tb1_idx_id2 on tb1(id);

reindex index tb1_idx_id;

create table tbref(id int, c1 int);
select create_refence_table('tbref');

postgres=# insert into tbref values(1,1),(2,2);
INSERT 0 2
postgres=# update tbref set c1=10;
UPDATE 2
postgres=# delete from tbref where id=1;
DELETE 1
postgres=# select * from tbref;
 id | c1 
----+----
  2 | 10
(1 row)

postgres=# truncate tb1;
TRUNCATE TABLE
postgres=# truncate tbref;
TRUNCATE TABLE

create table tbserial(id int,c1 int);
select create_distributed_table('tbserial','id');

postgres=# \d tbserial
                            Table "public.tbserial"
 Column |  Type   | Collation | Nullable |               Default                
--------+---------+-----------+----------+--------------------------------------
 id     | integer |           | not null | nextval('tbserial_id_seq'::regclass)
 c1     | integer |           |          | 

全局序列值 = groupid,节点内的顺序递增值

postgres=# \d tbserial_id_seq
                  Sequence "public.tbserial_id_seq"
  Type   | Start | Minimum |  Maximum   | Increment | Cycles? | Cache 
---------+-------+---------+------------+-----------+---------+-------
 integer |     1 |       1 | 2147483647 |         1 | no      |     1
Owned by: public.tbserial.id

postgres=# \d tbserial_id_seq
                    Sequence "public.tbserial_id_seq"
  Type  |   Start   |  Minimum  |  Maximum  | Increment | Cycles? | Cache 
--------+-----------+-----------+-----------+-----------+---------+-------
 bigint | 268435457 | 268435457 | 536870913 |         1 | no      |     1

postgres=# select * from pg_dist_node;
 nodeid | groupid | nodename  | nodeport | noderack | hasmetadata | isactive | noderole | nodecluster | metadatasynced | shouldhaveshards 
--------+---------+-----------+----------+----------+-------------+----------+----------+-------------+----------------+------------------
      2 |       2 | 127.0.0.1 |     9002 | default  | t           | t        | primary  | default     | t              | t
      1 |       1 | 127.0.0.1 |     9001 | default  | t           | t        | primary  | default     | t              | t
(2 rows)

postgres=# select * from pg_dist_local_group;
 groupid 
---------
       1
(1 row)

 - Test 1: HammerDB test with 250 users, 1,000,000 transactions per. 8 Node Citus MX 
          - (a) With local execution: `System achieved 116473 PostgreSQL TPM at 160355 NOPM`   
          - (b) without local execution: ` System achieved 61392 PostgreSQL TPM at 100503 NOPM`

  - Test 2: HammerDB test with 250 users, 10,000,000 transactions per. 8 Node Citus MX 
           - (a) With local execution: `System achieved 91921 PostgreSQL TPM at 174557 NOPM`   
           - (b) without local execution: ` System achieved 84186 PostgreSQL TPM at 98408 NOPM`

- Test 3: Pgbench, 1 worker node, -c64 -c256 -T 120
            - (a) Local execution enabled (tps): `select-only`: 56202   `simple-update`:  11771 `tpcb-like`: 7796
            - (a) Local execution disabled (tps): `select-only`:  24524 `simple-update`: 5077  `tpcb-like`:   3510 (some connection errors for tpcb-like)

postgres=# alter system set citus.max_shared_pool_size to 4;
ALTER SYSTEM
postgres=# select pg_reload_conf();
 pg_reload_conf 
----------------
 t
(1 row)
postgres=# begin;
BEGIN
postgres=*# update tb1 set c1=11;
UPDATE 1
postgres=*# select * from citus_remote_connection_stats();
 hostname  | port | database_name | connection_count_to_node 
-----------+------+---------------+--------------------------
 127.0.0.1 | 9002 | postgres      |                        4
 127.0.0.1 | 9001 | postgres      |                        4
(2 rows)

postgres=# select * from citus_remote_connection_stats();
 hostname  | port | database_name | connection_count_to_node 
-----------+------+---------------+--------------------------
 127.0.0.1 | 9002 | postgres      |                        2
 127.0.0.1 | 9001 | postgres      |                        2
(2 rows)

postgres=# set citus.executor_slow_start_interval='10ms';
SET
postgres=# begin;
BEGIN
postgres=*# update tb1 set c1=100;
UPDATE 0
postgres=*# select * from citus_remote_connection_stats();
 hostname  | port | database_name | connection_count_to_node 
-----------+------+---------------+--------------------------
 127.0.0.1 | 9002 | postgres      |                        4
 127.0.0.1 | 9001 | postgres      |                        4
(2 rows)

postgres=# set citus.executor_slow_start_interval='500ms';
SET
postgres=# begin;
BEGIN
postgres=*# update tb1 set c1=100;
UPDATE 0
postgres=*# select * from citus_remote_connection_stats();
 hostname  | port | database_name | connection_count_to_node 
-----------+------+---------------+--------------------------
 127.0.0.1 | 9002 | postgres      |                        2
 127.0.0.1 | 9001 | postgres      |                        2
(2 rows)

postgres=# set citus.executor_slow_start_interval = '0ms';
SET
postgres=# begin;
BEGIN
postgres=*# update tb1 set c1=100;
UPDATE 0
postgres=*# select * from citus_remote_connection_stats();
 hostname  | port | database_name | connection_count_to_node 
-----------+------+---------------+--------------------------
 127.0.0.1 | 9002 | postgres      |                        5
 127.0.0.1 | 9001 | postgres      |                       14
(2 rows)

static void
ManageWorkerPool(WorkerPool *workerPool)
{
...
		/* cannot open more than targetPoolSize connections */
		int maxNewConnectionCount = targetPoolSize - initiatedConnectionCount;//targetPoolSize的值为max(1,`citus.max_cached_conns_per_worker`)

		/* total number of connections that are (almost) available for tasks */
		int usableConnectionCount = UsableConnectionCount(workerPool);

		/*
		 * Number of additional connections we would need to run all ready tasks in
		 * parallel.
		 */
		int newConnectionsForReadyTasks = readyTaskCount - usableConnectionCount;

		/*
		 * Open enough connections to handle all tasks that are ready, but no more
		 * than the target pool size.
		 */
		newConnectionCount = Min(newConnectionsForReadyTasks, maxNewConnectionCount);

		if (newConnectionCount > 0 && ExecutorSlowStartInterval != SLOW_START_DISABLED)
		{
			if (MillisecondsPassedSince(workerPool->lastConnectionOpenTime) >=
				ExecutorSlowStartInterval)
			{
				newConnectionCount = Min(newConnectionCount,
										 workerPool->maxNewConnectionsPerCycle);

				/* increase the open rate every cycle (like TCP slow start) */
				workerPool->maxNewConnectionsPerCycle += 1;
			}
			else
			{
				/* wait a bit until opening more connections */
				return;
			}
		}

create table tb1(id int, c1 int);
select create_distributed_table('tb1','id');
set citus.shard_count to 16;
create table tb2(id int primary key, c1 int);
select create_distributed_table('tb2','id');

postgres=# explain INSERT INTO tb2
  SELECT * from tb1
  ON CONFLICT(id) DO UPDATE SET c1 = EXCLUDED.c1;
                                     QUERY PLAN                                     
------------------------------------------------------------------------------------
 Custom Scan (Citus INSERT ... SELECT)  (cost=0.00..0.00 rows=0 width=0)
   INSERT/SELECT method: repartition
   ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=100000 width=8)
         Task Count: 32
         Tasks Shown: One of 32
         ->  Task
               Node: host=127.0.0.1 port=9001 dbname=postgres
               ->  Seq Scan on tb1_102339 tb1  (cost=0.00..32.60 rows=2260 width=8)
(8 rows)

postgres=# set citus.task_assignment_policy TO "round-robin";
SET
postgres=# explain select * from tbref;
                                    QUERY PLAN                                    
----------------------------------------------------------------------------------
 Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
   Task Count: 1
   Tasks Shown: All
   ->  Task
         Node: host=127.0.0.1 port=9001 dbname=postgres
         ->  Seq Scan on tbref_102371 tbref  (cost=0.00..32.60 rows=2260 width=8)
(6 rows)

postgres=# explain select * from tbref;
                                    QUERY PLAN                                    
----------------------------------------------------------------------------------
 Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0)
   Task Count: 1
   Tasks Shown: All
   ->  Task
         Node: host=127.0.0.1 port=9002 dbname=postgres
         ->  Seq Scan on tbref_102371 tbref  (cost=0.00..32.60 rows=2260 width=8)
(6 rows)

COPY table tb1 to STDOUT

SELECT master_set_node_property('127.0.0.1', 9002, 'shouldhaveshards', false);

postgres=# \df master_update_node
List of functions
-[ RECORD 1 ]-------+-----------------------------------------------------------------------------------------------------------------------------
Schema              | pg_catalog
Name                | master_update_node
Result data type    | void
Argument data types | node_id integer, new_node_name text, new_node_port integer, force boolean DEFAULT false, lock_cooldown integer DEFAULT 10000
Type                | func

create table tba(id int,c1 int);
select create_distributed_table('tba','id');
create table tbb(id int,c1 int);
select create_distributed_table('tbb','id');
create table tbc(id text,c1 int);
select create_distributed_table('tbc','id');

postgres=# select * from pg_dist_partition where logicalrelid in ('tba'::regclass,'tbb'::regclass);
 logicalrelid | partmethod |                                                        partkey                                                         | colocationid | repmodel 
--------------+------------+------------------------------------------------------------------------------------------------------------------------+--------------+----------
 tba          | h          | {VAR :varno 1 :varattno 1 :vartype 23 :vartypmod -1 :varcollid 0 :varlevelsup 0 :varnoold 1 :varoattno 1 :location -1} |            3 | s
 tbb          | h          | {VAR :varno 1 :varattno 1 :vartype 23 :vartypmod -1 :varcollid 0 :varlevelsup 0 :varnoold 1 :varoattno 1 :location -1} |            3 | s
(2 rows)

postgres=# SELECT update_distributed_table_colocation('tbb', colocate_with => 'none');
 update_distributed_table_colocation 
-------------------------------------
 
(1 row)

postgres=# select * from pg_dist_partition where logicalrelid in ('tba'::regclass,'tbb'::regclass);
 logicalrelid | partmethod |                                                        partkey                                                         | colocationid | repmodel 
--------------+------------+------------------------------------------------------------------------------------------------------------------------+--------------+----------
 tba          | h          | {VAR :varno 1 :varattno 1 :vartype 23 :vartypmod -1 :varcollid 0 :varlevelsup 0 :varnoold 1 :varoattno 1 :location -1} |            3 | s
 tbb          | h          | {VAR :varno 1 :varattno 1 :vartype 23 :vartypmod -1 :varcollid 0 :varlevelsup 0 :varnoold 1 :varoattno 1 :location -1} |           14 | s
(2 rows)

postgres=# SELECT update_distributed_table_colocation('tbb', colocate_with => 'tba');
 update_distributed_table_colocation 
-------------------------------------
 
(1 row)

postgres=# select * from pg_dist_partition where logicalrelid in ('tba'::regclass,'tbb'::regclass);
 logicalrelid | partmethod |                                                        partkey                                                         | colocationid | repmodel 
--------------+------------+------------------------------------------------------------------------------------------------------------------------+--------------+----------
 tba          | h          | {VAR :varno 1 :varattno 1 :vartype 23 :vartypmod -1 :varcollid 0 :varlevelsup 0 :varnoold 1 :varoattno 1 :location -1} |            3 | s
 tbb          | h          | {VAR :varno 1 :varattno 1 :vartype 23 :vartypmod -1 :varcollid 0 :varlevelsup 0 :varnoold 1 :varoattno 1 :location -1} |            3 | s
(2 rows)

postgres=# SELECT mark_tables_colocated('tba', ARRAY['tbb', 'tbc']);
ERROR:  cannot colocate tables tba and tbc
DETAIL:  Distribution column types don't match for tba and tbc.

postgres=# SELECT mark_tables_colocated('tba', ARRAY['tbb']);
 mark_tables_colocated 
-----------------------
 
(1 row)

SELECT truncate_local_data_after_distributing_table('tb1');

citus.replicate_reference_tables_on_activate = off

select citus_create_restore_point('foo');

postgres=# show citus.node_conninfo;
 citus.node_conninfo 
---------------------
 sslmode=prefer
(1 row)

postgres=# show citus.enable_statistics_collection;
 citus.enable_statistics_collection 
------------------------------------
 off
(1 row)

postgres=# select * from citus_remote_connection_stats();
 hostname  | port | database_name | connection_count_to_node 
-----------+------+---------------+--------------------------
 127.0.0.1 | 9002 | postgres      |                        3
 127.0.0.1 | 9001 | postgres      |                        3
(2 rows)

postgres=# select * from citus_dist_stat_activity;
-[ RECORD 1 ]----------+------------------------------
query_hostname         | coordinator_host
query_hostport         | 9000
master_query_host_name | coordinator_host
master_query_host_port | 9000
transaction_number     | 57
transaction_stamp      | 2020-06-19 15:05:22.142242+08
datid                  | 13593
datname                | postgres
pid                    | 2574
usesysid               | 10
usename                | postgres
application_name       | psql
client_addr            | 
client_hostname        | 
client_port            | -1
backend_start          | 2020-06-19 10:57:58.472994+08
xact_start             | 2020-06-19 15:05:17.45487+08
query_start            | 2020-06-19 15:05:22.140954+08
state_change           | 2020-06-19 15:05:22.140957+08
wait_event_type        | Client
wait_event             | ClientRead
state                  | active
backend_xid            | 
backend_xmin           | 5114
query                  | select * from tb1;
backend_type           | client backend

 *  An important note on this views is that they only show the activity
 *  that are inside distributed transactions. Distributed transactions
 *  cover the following:
 *     - All multi-shard modifications (DDLs, COPY, UPDATE, DELETE, INSERT .. SELECT)
 *     - All multi-shard queries with CTEs (modifying CTEs, read-only CTEs)
 *     - All recursively planned subqueries
 *     - All queries within transaction blocks (BEGIN; query; COMMMIT;)
 *
 *  In other words, the following types of queries won't be observed in these
 *  views:
 *      - Single-shard queries that are not inside transaction blocks
 *      - Multi-shard select queries that are not inside transaction blocks
 *      - Task-tracker queries

postgres=# select * from citus_worker_stat_activity where transaction_number = 57;
-[ RECORD 1 ]----------+---------------------------------------------
query_hostname         | 127.0.0.1
query_hostport         | 9001
master_query_host_name | coordinator_host
master_query_host_port | 9000
transaction_number     | 57
transaction_stamp      | 2020-06-19 15:05:22.142242+08
datid                  | 13593
datname                | postgres
pid                    | 4108
usesysid               | 10
usename                | postgres
application_name       | citus
client_addr            | 127.0.0.1
client_hostname        | 
client_port            | 33676
backend_start          | 2020-06-19 15:05:22.162829+08
xact_start             | 2020-06-19 15:05:22.168811+08
query_start            | 2020-06-19 15:05:22.171398+08
state_change           | 2020-06-19 15:05:22.172237+08
wait_event_type        | Client
wait_event             | ClientRead
state                  | idle in transaction
backend_xid            | 
backend_xmin           | 
query                  | SELECT id, c1 FROM tb1_102369 tb1 WHERE true
backend_type           | client backend
...

CREATE TABLE numbers AS
  SELECT i, 0 AS j FROM generate_series(1,10) AS i;
SELECT create_distributed_table('numbers', 'i');

-- session 1                           -- session 2
-------------------------------------  -------------------------------------
BEGIN;
UPDATE numbers SET j = 2 WHERE i = 1;
                                       BEGIN;
                                       UPDATE numbers SET j = 3 WHERE i = 1;
                                       -- (this blocks)

SELECT * FROM citus_lock_waits;
-[ RECORD 1 ]-------------------------+----------------------------------------
waiting_pid                           | 88624
blocking_pid                          | 88615
blocked_statement                     | UPDATE numbers SET j = 3 WHERE i = 1;
current_statement_in_blocking_process | UPDATE numbers SET j = 2 WHERE i = 1;
waiting_node_id                       | 0
blocking_node_id                      | 0
waiting_node_name                     | coordinator_host
blocking_node_name                    | coordinator_host
waiting_node_port                     | 5432
blocking_node_port                    | 5432

postgres=# select * from master_get_table_metadata('tb1');
-[ RECORD 1 ]---------+-----------
logical_relid         | 17148
part_storage_type     | t
part_method           | h
part_key              | id
part_replica_count    | 1
part_max_size         | 1073741824
part_placement_policy | 2

postgres=# SELECT get_shard_id_for_distribution_column('tb1', 4);
 get_shard_id_for_distribution_column 
--------------------------------------
                               102347
(1 row)

create table tbl(id int,c1 int);
select create_distributed_table('tbserial','id');

postgres=# insert into tb1 values(3,3);
ERROR:  writing to worker nodes is not currently allowed
DETAIL:  the database is in recovery mode 
postgres=# set citus.writable_standby_coordinator TO ON;
SET
postgres=# insert into tb1 values(3,3);
INSERT 0 1
postgres=# update tb1 set c1=20 where id=3;
UPDATE 1
postgres=# delete from tb1 where id=3;
DELETE 1

postgres=# update tb1 set c1=20;
ERROR:  cannot assign TransactionIds during recovery
postgres=# delete from tb1 where c1=20;
ERROR:  cannot assign TransactionIds during recovery

postgres=# begin;
BEGIN
postgres=*# insert into tb1 values(3,3);
INSERT 0 1
postgres=*# insert into tb1 values(4,4);
INSERT 0 1
postgres=*# commit;
ERROR:  cannot assign TransactionIds during recovery

postgres=# set citus.multi_shard_commit_protocol TO '1pc';
SET
postgres=# begin;
BEGIN
postgres=*# insert into tb1 values(4,4);
INSERT 0 1
postgres=*# insert into tb1 values(5,5);
INSERT 0 1
postgres=*# commit;

postgres=# set citus.multi_shard_commit_protocol TO '1pc';
SET
postgres=# update tb1 set c1=10;
UPDATE 3