mirror of https://github.com/citusdata/citus.git
1026 lines
40 KiB
Plaintext
1026 lines
40 KiB
Plaintext
--
|
|
-- MULTI_COPY
|
|
--
|
|
ALTER SEQUENCE pg_catalog.pg_dist_shardid_seq RESTART 560000;
|
|
-- Create a new hash-partitioned table into which to COPY
|
|
CREATE TABLE customer_copy_hash (
|
|
c_custkey integer,
|
|
c_name varchar(25) not null,
|
|
c_address varchar(40),
|
|
c_nationkey integer,
|
|
c_phone char(15),
|
|
c_acctbal decimal(15,2),
|
|
c_mktsegment char(10),
|
|
c_comment varchar(117),
|
|
primary key (c_custkey));
|
|
SET citus.shard_replication_factor TO 1;
|
|
SELECT create_distributed_table('customer_copy_hash', 'c_custkey', shard_count:=64);
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
-- Test empty copy
|
|
COPY customer_copy_hash FROM STDIN;
|
|
-- Test syntax error
|
|
COPY customer_copy_hash (c_custkey,c_name) FROM STDIN;
|
|
ERROR: invalid input syntax for integer: "1,customer1"
|
|
CONTEXT: COPY customer_copy_hash, line 1, column c_custkey: "1,customer1"
|
|
-- Test invalid option
|
|
COPY customer_copy_hash (c_custkey,c_name) FROM STDIN (append_to_shard xxxxx);
|
|
ERROR: append_to_shard is only valid for append-distributed tables
|
|
-- Confirm that no data was copied
|
|
SELECT count(*) FROM customer_copy_hash;
|
|
count
|
|
---------------------------------------------------------------------
|
|
0
|
|
(1 row)
|
|
|
|
-- Test primary key violation
|
|
COPY customer_copy_hash (c_custkey, c_name) FROM STDIN
|
|
WITH (FORMAT 'csv');
|
|
ERROR: duplicate key value violates unique constraint "customer_copy_hash_pkey_560048"
|
|
DETAIL: Key (c_custkey)=(2) already exists.
|
|
-- Confirm that no data was copied
|
|
SELECT count(*) FROM customer_copy_hash;
|
|
count
|
|
---------------------------------------------------------------------
|
|
0
|
|
(1 row)
|
|
|
|
-- Test headers option
|
|
COPY customer_copy_hash (c_custkey, c_name) FROM STDIN
|
|
WITH (FORMAT 'csv', HEADER true, FORCE_NULL (c_custkey));
|
|
-- Confirm that only first row was skipped
|
|
SELECT count(*) FROM customer_copy_hash;
|
|
count
|
|
---------------------------------------------------------------------
|
|
3
|
|
(1 row)
|
|
|
|
-- Test force_not_null option
|
|
COPY customer_copy_hash (c_custkey, c_name, c_address) FROM STDIN
|
|
WITH (FORMAT 'csv', QUOTE '"', FORCE_NOT_NULL (c_address));
|
|
-- Confirm that value is not null
|
|
SELECT count(c_address) FROM customer_copy_hash WHERE c_custkey = 4;
|
|
count
|
|
---------------------------------------------------------------------
|
|
1
|
|
(1 row)
|
|
|
|
-- Test force_null option
|
|
COPY customer_copy_hash (c_custkey, c_name, c_address) FROM STDIN
|
|
WITH (FORMAT 'csv', QUOTE '"', FORCE_NULL (c_address));
|
|
-- Confirm that value is null
|
|
SELECT count(c_address) FROM customer_copy_hash WHERE c_custkey = 5;
|
|
count
|
|
---------------------------------------------------------------------
|
|
0
|
|
(1 row)
|
|
|
|
-- Test null violation
|
|
COPY customer_copy_hash (c_custkey, c_name) FROM STDIN
|
|
WITH (FORMAT 'csv');
|
|
ERROR: null value in column "c_name" violates not-null constraint
|
|
DETAIL: Failing row contains (8, null, null, null, null, null, null, null).
|
|
-- Confirm that no data was copied
|
|
SELECT count(*) FROM customer_copy_hash;
|
|
count
|
|
---------------------------------------------------------------------
|
|
5
|
|
(1 row)
|
|
|
|
-- Test server-side copy from program
|
|
COPY customer_copy_hash (c_custkey, c_name) FROM PROGRAM 'echo 9 customer9'
|
|
WITH (DELIMITER ' ');
|
|
-- Confirm that data was copied
|
|
SELECT count(*) FROM customer_copy_hash WHERE c_custkey = 9;
|
|
count
|
|
---------------------------------------------------------------------
|
|
1
|
|
(1 row)
|
|
|
|
-- Test server-side copy from file
|
|
COPY customer_copy_hash FROM '@abs_srcdir@/data/customer.2.data' WITH (DELIMITER '|');
|
|
-- Confirm that data was copied
|
|
SELECT count(*) FROM customer_copy_hash;
|
|
count
|
|
---------------------------------------------------------------------
|
|
1006
|
|
(1 row)
|
|
|
|
-- Test client-side copy from file
|
|
\copy customer_copy_hash FROM '@abs_srcdir@/data/customer.3.data' WITH (DELIMITER '|');
|
|
-- Confirm that data was copied
|
|
SELECT count(*) FROM customer_copy_hash;
|
|
count
|
|
---------------------------------------------------------------------
|
|
2006
|
|
(1 row)
|
|
|
|
-- Update shard statistics for hash-partitioned table
|
|
SELECT citus_update_shard_statistics(560000);
|
|
citus_update_shard_statistics
|
|
---------------------------------------------------------------------
|
|
8192
|
|
(1 row)
|
|
|
|
SELECT shardid, shardlength FROM pg_dist_shard_placement WHERE shardid = 560000;
|
|
shardid | shardlength
|
|
---------------------------------------------------------------------
|
|
560000 | 8192
|
|
(1 row)
|
|
|
|
-- Create a new hash-partitioned table with default now() function
|
|
CREATE TABLE customer_with_default(
|
|
c_custkey integer,
|
|
c_name varchar(25) not null,
|
|
c_time timestamp default now());
|
|
SET citus.shard_replication_factor TO 1;
|
|
SELECT create_distributed_table('customer_with_default', 'c_custkey', shard_count:=64);
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
-- Test with default values for now() function
|
|
COPY customer_with_default (c_custkey, c_name) FROM STDIN
|
|
WITH (FORMAT 'csv');
|
|
-- Confirm that data was copied with now() function
|
|
SELECT count(*) FROM customer_with_default where c_time IS NOT NULL;
|
|
count
|
|
---------------------------------------------------------------------
|
|
2
|
|
(1 row)
|
|
|
|
-- Add columns to the table and perform a COPY
|
|
ALTER TABLE customer_copy_hash ADD COLUMN extra1 INT DEFAULT 0;
|
|
ALTER TABLE customer_copy_hash ADD COLUMN extra2 INT DEFAULT 0;
|
|
COPY customer_copy_hash (c_custkey, c_name, extra1, extra2) FROM STDIN CSV;
|
|
SELECT * FROM customer_copy_hash WHERE extra1 = 1;
|
|
c_custkey | c_name | c_address | c_nationkey | c_phone | c_acctbal | c_mktsegment | c_comment | extra1 | extra2
|
|
---------------------------------------------------------------------
|
|
10 | customer10 | | | | | | | 1 | 5
|
|
(1 row)
|
|
|
|
-- Test dropping an intermediate column
|
|
ALTER TABLE customer_copy_hash DROP COLUMN extra1;
|
|
COPY customer_copy_hash (c_custkey, c_name, extra2) FROM STDIN CSV;
|
|
SELECT * FROM customer_copy_hash WHERE c_custkey = 11;
|
|
c_custkey | c_name | c_address | c_nationkey | c_phone | c_acctbal | c_mktsegment | c_comment | extra2
|
|
---------------------------------------------------------------------
|
|
11 | customer11 | | | | | | | 5
|
|
(1 row)
|
|
|
|
-- Test dropping the last column
|
|
ALTER TABLE customer_copy_hash DROP COLUMN extra2;
|
|
COPY customer_copy_hash (c_custkey, c_name) FROM STDIN CSV;
|
|
SELECT * FROM customer_copy_hash WHERE c_custkey = 12;
|
|
c_custkey | c_name | c_address | c_nationkey | c_phone | c_acctbal | c_mktsegment | c_comment
|
|
---------------------------------------------------------------------
|
|
12 | customer12 | | | | | |
|
|
(1 row)
|
|
|
|
-- Create a new range-partitioned table into which to COPY
|
|
CREATE TABLE customer_copy_range (
|
|
c_custkey integer,
|
|
c_name varchar(25),
|
|
c_address varchar(40),
|
|
c_nationkey integer,
|
|
c_phone char(15),
|
|
c_acctbal decimal(15,2),
|
|
c_mktsegment char(10),
|
|
c_comment varchar(117),
|
|
primary key (c_custkey));
|
|
SELECT master_create_distributed_table('customer_copy_range', 'c_custkey', 'range');
|
|
master_create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
-- Test COPY into empty range-partitioned table
|
|
COPY customer_copy_range FROM '@abs_srcdir@/data/customer.1.data' WITH (DELIMITER '|');
|
|
ERROR: could not find any shards into which to copy
|
|
DETAIL: No shards exist for distributed table "customer_copy_range".
|
|
SELECT master_create_empty_shard('customer_copy_range') AS new_shard_id
|
|
\gset
|
|
UPDATE pg_dist_shard SET shardminvalue = 1, shardmaxvalue = 500
|
|
WHERE shardid = :new_shard_id;
|
|
SELECT master_create_empty_shard('customer_copy_range') AS new_shard_id
|
|
\gset
|
|
UPDATE pg_dist_shard SET shardminvalue = 501, shardmaxvalue = 1000
|
|
WHERE shardid = :new_shard_id;
|
|
-- Test copy into range-partitioned table
|
|
COPY customer_copy_range FROM '@abs_srcdir@/data/customer.1.data' WITH (DELIMITER '|');
|
|
-- Check whether data went into the right shard (maybe)
|
|
SELECT min(c_custkey), max(c_custkey), avg(c_custkey), count(*)
|
|
FROM customer_copy_range WHERE c_custkey <= 500;
|
|
min | max | avg | count
|
|
---------------------------------------------------------------------
|
|
1 | 500 | 250.5000000000000000 | 500
|
|
(1 row)
|
|
|
|
-- Check whether data was copied
|
|
SELECT count(*) FROM customer_copy_range;
|
|
count
|
|
---------------------------------------------------------------------
|
|
1000
|
|
(1 row)
|
|
|
|
SELECT shardid, shardlength FROM pg_dist_shard_placement WHERE shardid = :new_shard_id;
|
|
shardid | shardlength
|
|
---------------------------------------------------------------------
|
|
560129 | 0
|
|
(1 row)
|
|
|
|
-- Update shard statistics for range-partitioned shard
|
|
SELECT citus_update_shard_statistics(:new_shard_id);
|
|
citus_update_shard_statistics
|
|
---------------------------------------------------------------------
|
|
131072
|
|
(1 row)
|
|
|
|
SELECT shardid, shardlength FROM pg_dist_shard_placement WHERE shardid = :new_shard_id;
|
|
shardid | shardlength
|
|
---------------------------------------------------------------------
|
|
560129 | 131072
|
|
(1 row)
|
|
|
|
-- Create a new append-partitioned table into which to COPY
|
|
CREATE TABLE customer_copy_append (
|
|
c_custkey integer,
|
|
c_name varchar(25) not null,
|
|
c_address varchar(40),
|
|
c_nationkey integer,
|
|
c_phone char(15),
|
|
c_acctbal decimal(15,2),
|
|
c_mktsegment char(10),
|
|
c_comment varchar(117));
|
|
SELECT create_distributed_table('customer_copy_append', 'c_custkey', 'append');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SET citus.shard_replication_factor TO 2;
|
|
-- Test syntax error
|
|
BEGIN;
|
|
SELECT master_create_empty_shard('customer_copy_append') AS shardid \gset
|
|
COPY customer_copy_append(c_custkey, c_name) FROM STDIN WITH (FORMAT 'csv', append_to_shard :shardid);
|
|
ERROR: invalid input syntax for integer: "notinteger"
|
|
CONTEXT: COPY customer_copy_append, line 3, column c_custkey: "notinteger"
|
|
END;
|
|
-- Test that no shard is created for failing copy
|
|
SELECT count(*) FROM pg_dist_shard WHERE logicalrelid = 'customer_copy_append'::regclass;
|
|
count
|
|
---------------------------------------------------------------------
|
|
0
|
|
(1 row)
|
|
|
|
-- Test empty copy
|
|
BEGIN;
|
|
SELECT master_create_empty_shard('customer_copy_append') AS shardid \gset
|
|
COPY customer_copy_append FROM STDIN WITH (append_to_shard :shardid);
|
|
END;
|
|
-- Test that a shard is created
|
|
SELECT count(*) FROM pg_dist_shard WHERE logicalrelid = 'customer_copy_append'::regclass;
|
|
count
|
|
---------------------------------------------------------------------
|
|
1
|
|
(1 row)
|
|
|
|
-- Test proper copy
|
|
BEGIN;
|
|
SELECT master_create_empty_shard('customer_copy_append') AS shardid \gset
|
|
COPY customer_copy_append(c_custkey, c_name) FROM STDIN WITH (FORMAT 'csv', append_to_shard :shardid);
|
|
END;
|
|
-- Check whether data was copied properly
|
|
SELECT * FROM customer_copy_append;
|
|
c_custkey | c_name | c_address | c_nationkey | c_phone | c_acctbal | c_mktsegment | c_comment
|
|
---------------------------------------------------------------------
|
|
1 | customer1 | | | | | |
|
|
2 | customer2 | | | | | |
|
|
(2 rows)
|
|
|
|
-- Manipulate manipulate and check shard statistics for append-partitioned table shard
|
|
UPDATE pg_dist_shard_placement SET shardlength = 0 WHERE shardid = 560132;
|
|
SELECT shardid, shardlength FROM pg_dist_shard_placement WHERE shardid = 560132;
|
|
shardid | shardlength
|
|
---------------------------------------------------------------------
|
|
560132 | 0
|
|
560132 | 0
|
|
(2 rows)
|
|
|
|
-- Update shard statistics for append-partitioned shard
|
|
SELECT citus_update_shard_statistics(560132);
|
|
citus_update_shard_statistics
|
|
---------------------------------------------------------------------
|
|
8192
|
|
(1 row)
|
|
|
|
SELECT shardid, shardlength FROM pg_dist_shard_placement WHERE shardid = 560132;
|
|
shardid | shardlength
|
|
---------------------------------------------------------------------
|
|
560132 | 8192
|
|
560132 | 8192
|
|
(2 rows)
|
|
|
|
-- Create lineitem table
|
|
CREATE TABLE lineitem_copy_append (
|
|
l_orderkey bigint not null,
|
|
l_partkey integer not null,
|
|
l_suppkey integer not null,
|
|
l_linenumber integer not null,
|
|
l_quantity decimal(15, 2) not null,
|
|
l_extendedprice decimal(15, 2) not null,
|
|
l_discount decimal(15, 2) not null,
|
|
l_tax decimal(15, 2) not null,
|
|
l_returnflag char(1) not null,
|
|
l_linestatus char(1) not null,
|
|
l_shipdate date not null,
|
|
l_commitdate date not null,
|
|
l_receiptdate date not null,
|
|
l_shipinstruct char(25) not null,
|
|
l_shipmode char(10) not null,
|
|
l_comment varchar(44) not null);
|
|
SELECT create_distributed_table('lineitem_copy_append', 'l_orderkey', 'append');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
BEGIN;
|
|
SELECT master_create_empty_shard('lineitem_copy_append') AS shardid \gset
|
|
COPY lineitem_copy_append FROM '@abs_srcdir@/data/lineitem.1.data' with (delimiter '|', append_to_shard :shardid);
|
|
END;
|
|
SELECT count(*) FROM pg_dist_shard WHERE logicalrelid = 'lineitem_copy_append'::regclass;
|
|
count
|
|
---------------------------------------------------------------------
|
|
1
|
|
(1 row)
|
|
|
|
-- trigger some errors on the append_to_shard option
|
|
COPY lineitem_copy_append FROM '@abs_srcdir@/data/lineitem.1.data' with (delimiter '|', append_to_shard xxxxx);
|
|
ERROR: could not find valid entry for shard xxxxx
|
|
COPY lineitem_copy_append FROM '@abs_srcdir@/data/lineitem.1.data' with (delimiter '|', append_to_shard xxxxx);
|
|
ERROR: shard xxxxx does not belong to table lineitem_copy_append
|
|
-- Test schema support on append partitioned tables
|
|
CREATE SCHEMA append;
|
|
CREATE TABLE append.customer_copy (
|
|
c_custkey integer ,
|
|
c_name varchar(25) not null,
|
|
c_address varchar(40),
|
|
c_nationkey integer,
|
|
c_phone char(15),
|
|
c_acctbal decimal(15,2),
|
|
c_mktsegment char(10),
|
|
c_comment varchar(117));
|
|
SELECT create_distributed_table('append.customer_copy', 'c_custkey', 'append');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SELECT master_create_empty_shard('append.customer_copy') AS shardid1 \gset
|
|
SELECT master_create_empty_shard('append.customer_copy') AS shardid2 \gset
|
|
-- Test copy from the master node
|
|
COPY append.customer_copy FROM '@abs_srcdir@/data/customer.1.data' with (delimiter '|', append_to_shard :shardid1);
|
|
COPY append.customer_copy FROM '@abs_srcdir@/data/customer.2.data' with (delimiter '|', append_to_shard :shardid2);
|
|
-- Test the content of the table
|
|
SELECT min(c_custkey), max(c_custkey), avg(c_acctbal), count(*) FROM append.customer_copy;
|
|
min | max | avg | count
|
|
---------------------------------------------------------------------
|
|
1 | 7000 | 4443.8028800000000000 | 2000
|
|
(1 row)
|
|
|
|
-- Test with table name which contains special character
|
|
CREATE TABLE "customer_with_special_\\_character"(
|
|
c_custkey integer,
|
|
c_name varchar(25) not null);
|
|
SELECT master_create_distributed_table('"customer_with_special_\\_character"', 'c_custkey', 'hash');
|
|
master_create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SELECT master_create_worker_shards('"customer_with_special_\\_character"', 4, 1);
|
|
master_create_worker_shards
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
COPY "customer_with_special_\\_character" (c_custkey, c_name) FROM STDIN
|
|
WITH (FORMAT 'csv');
|
|
-- Confirm that data was copied
|
|
SELECT count(*) FROM "customer_with_special_\\_character";
|
|
count
|
|
---------------------------------------------------------------------
|
|
2
|
|
(1 row)
|
|
|
|
-- Test with table name which starts with number
|
|
CREATE TABLE "1_customer"(
|
|
c_custkey integer,
|
|
c_name varchar(25) not null);
|
|
SELECT master_create_distributed_table('"1_customer"', 'c_custkey', 'hash');
|
|
master_create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SELECT master_create_worker_shards('"1_customer"', 4, 1);
|
|
master_create_worker_shards
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
COPY "1_customer" (c_custkey, c_name) FROM STDIN
|
|
WITH (FORMAT 'csv');
|
|
-- Confirm that data was copied
|
|
SELECT count(*) FROM "1_customer";
|
|
count
|
|
---------------------------------------------------------------------
|
|
2
|
|
(1 row)
|
|
|
|
-- Test COPY with types having different Oid at master and workers
|
|
CREATE TYPE number_pack AS (
|
|
number1 integer,
|
|
number2 integer
|
|
);
|
|
CREATE TYPE super_number_pack AS (
|
|
packed_number1 number_pack,
|
|
packed_number2 number_pack
|
|
);
|
|
-- Test array of user-defined type with hash distribution
|
|
CREATE TABLE packed_numbers_hash (
|
|
id integer,
|
|
packed_numbers number_pack[]
|
|
);
|
|
SELECT master_create_distributed_table('packed_numbers_hash', 'id', 'hash');
|
|
master_create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SELECT master_create_worker_shards('packed_numbers_hash', 4, 1);
|
|
master_create_worker_shards
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
COPY (SELECT 1, ARRAY[ROW(42, 42), ROW(42, 42)]) TO :'temp_dir''copy_test_array_of_composite';
|
|
COPY packed_numbers_hash FROM :'temp_dir''copy_test_array_of_composite';
|
|
-- Verify data is actually copied
|
|
SELECT * FROM packed_numbers_hash;
|
|
id | packed_numbers
|
|
---------------------------------------------------------------------
|
|
1 | {"(42,42)","(42,42)"}
|
|
(1 row)
|
|
|
|
-- Test composite type containing an element with different Oid with hash distribution
|
|
CREATE TABLE super_packed_numbers_hash (
|
|
id integer,
|
|
super_packed_number super_number_pack
|
|
);
|
|
SELECT master_create_distributed_table('super_packed_numbers_hash', 'id', 'hash');
|
|
master_create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SELECT master_create_worker_shards('super_packed_numbers_hash', 4, 1);
|
|
master_create_worker_shards
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
COPY (SELECT 1, ROW(ROW(42, 42), ROW(42, 42))) TO :'temp_dir''copy_test_composite_of_composite';
|
|
COPY super_packed_numbers_hash FROM :'temp_dir''copy_test_composite_of_composite';
|
|
-- Verify data is actually copied
|
|
SELECT * FROM super_packed_numbers_hash;
|
|
id | super_packed_number
|
|
---------------------------------------------------------------------
|
|
1 | ("(42,42)","(42,42)")
|
|
(1 row)
|
|
|
|
-- Test array of user-defined type with append distribution
|
|
CREATE TABLE packed_numbers_append (
|
|
id integer,
|
|
packed_numbers number_pack[]
|
|
);
|
|
SELECT create_distributed_table('packed_numbers_append', 'id', 'append');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SELECT master_create_empty_shard('packed_numbers_append') AS shardid \gset
|
|
COPY packed_numbers_append FROM :'temp_dir''copy_test_array_of_composite' WITH (append_to_shard :shardid);
|
|
-- Verify data is actually copied
|
|
SELECT * FROM packed_numbers_append;
|
|
id | packed_numbers
|
|
---------------------------------------------------------------------
|
|
1 | {"(42,42)","(42,42)"}
|
|
(1 row)
|
|
|
|
-- Test composite type containing an element with different Oid with append distribution
|
|
CREATE TABLE super_packed_numbers_append (
|
|
id integer,
|
|
super_packed_number super_number_pack
|
|
);
|
|
SELECT create_distributed_table('super_packed_numbers_append', 'id', 'append');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SELECT master_create_empty_shard('super_packed_numbers_append') AS shardid \gset
|
|
COPY super_packed_numbers_append FROM :'temp_dir''copy_test_composite_of_composite' WITH (append_to_shard :shardid);
|
|
-- Verify data is actually copied
|
|
SELECT * FROM super_packed_numbers_append;
|
|
id | super_packed_number
|
|
---------------------------------------------------------------------
|
|
1 | ("(42,42)","(42,42)")
|
|
(1 row)
|
|
|
|
-- Test copy on append for composite type partition column
|
|
CREATE TABLE composite_partition_column_table(
|
|
id integer,
|
|
composite_column number_pack
|
|
);
|
|
SELECT create_distributed_table('composite_partition_column_table', 'composite_column', 'append');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SELECT master_create_empty_shard('composite_partition_column_table') AS shardid \gset
|
|
COPY composite_partition_column_table FROM STDIN WITH (FORMAT 'csv', append_to_shard :shardid);
|
|
-- Test copy on append distributed tables do not create shards on removed workers
|
|
SET citus.shard_replication_factor TO 2;
|
|
CREATE TABLE numbers_append (a int, b int);
|
|
SELECT create_distributed_table('numbers_append', 'a', 'append');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
-- no shards is created yet
|
|
SELECT shardid, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_append'::regclass order by placementid;
|
|
shardid | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
(0 rows)
|
|
|
|
SELECT master_create_empty_shard('numbers_append') AS shardid1 \gset
|
|
SELECT master_create_empty_shard('numbers_append') AS shardid2 \gset
|
|
COPY numbers_append FROM STDIN WITH (FORMAT 'csv', append_to_shard :shardid1);
|
|
COPY numbers_append FROM STDIN WITH (FORMAT 'csv', append_to_shard :shardid2);
|
|
-- verify there are shards at both workers
|
|
SELECT shardid, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_append'::regclass order by placementid;
|
|
shardid | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
560155 | localhost | 57637
|
|
560155 | localhost | 57638
|
|
560156 | localhost | 57638
|
|
560156 | localhost | 57637
|
|
(4 rows)
|
|
|
|
-- disable the first node
|
|
SET client_min_messages TO ERROR;
|
|
\set VERBOSITY terse
|
|
SELECT master_disable_node('localhost', :worker_1_port);
|
|
ERROR: cannot remove or disable the node localhost:xxxxx because because it contains the only shard placement for shard xxxxx
|
|
SELECT public.wait_until_metadata_sync(30000);
|
|
wait_until_metadata_sync
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
RESET client_min_messages;
|
|
\set VERBOSITY default
|
|
-- set replication factor to 1 so that copy will
|
|
-- succeed without replication count error
|
|
SET citus.shard_replication_factor TO 1;
|
|
-- add two new shards and verify they are created at the other node
|
|
SELECT master_create_empty_shard('numbers_append') AS shardid1 \gset
|
|
SELECT master_create_empty_shard('numbers_append') AS shardid2 \gset
|
|
COPY numbers_append FROM STDIN WITH (FORMAT 'csv', append_to_shard :shardid1);
|
|
COPY numbers_append FROM STDIN WITH (FORMAT 'csv', append_to_shard :shardid2);
|
|
SELECT shardid, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_append'::regclass order by placementid;
|
|
shardid | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
560155 | localhost | 57637
|
|
560155 | localhost | 57638
|
|
560156 | localhost | 57638
|
|
560156 | localhost | 57637
|
|
560157 | localhost | 57637
|
|
560158 | localhost | 57638
|
|
(6 rows)
|
|
|
|
-- add the node back
|
|
-- before adding the node, add pg_dist_object entry for tables created with
|
|
-- master_create_distributed_table as we don't have the entry for them.
|
|
INSERT INTO citus.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'objects'::regclass::oid, 0);
|
|
INSERT INTO citus.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'customer_with_special_\\_character'::regclass::oid, 0);
|
|
INSERT INTO citus.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, '1_customer'::regclass::oid, 0);
|
|
INSERT INTO citus.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'packed_numbers_hash'::regclass::oid, 0);
|
|
INSERT INTO citus.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'super_packed_numbers_hash'::regclass::oid, 0);
|
|
INSERT INTO citus.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'table_to_distribute'::regclass::oid, 0);
|
|
INSERT INTO citus.pg_dist_object(classid, objid, objsubid) values('pg_class'::regclass::oid, 'second_dustbunnies'::regclass::oid, 0);
|
|
SELECT 1 FROM master_activate_node('localhost', :worker_1_port);
|
|
NOTICE: Replicating postgres objects to node localhost:57637
|
|
DETAIL: There are 114 objects to replicate, depending on your environment this might take a while
|
|
?column?
|
|
---------------------------------------------------------------------
|
|
1
|
|
(1 row)
|
|
|
|
RESET client_min_messages;
|
|
RESET citus.shard_replication_factor;
|
|
-- add two new shards and verify they are created at both workers
|
|
SELECT master_create_empty_shard('numbers_append') AS shardid1 \gset
|
|
SELECT master_create_empty_shard('numbers_append') AS shardid2 \gset
|
|
COPY numbers_append FROM STDIN WITH (FORMAT 'csv', append_to_shard :shardid1);
|
|
COPY numbers_append FROM STDIN WITH (FORMAT 'csv', append_to_shard :shardid2);
|
|
SELECT shardid, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_append'::regclass order by placementid;
|
|
shardid | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
560155 | localhost | 57637
|
|
560155 | localhost | 57638
|
|
560156 | localhost | 57638
|
|
560156 | localhost | 57637
|
|
560157 | localhost | 57637
|
|
560158 | localhost | 57638
|
|
560159 | localhost | 57637
|
|
560159 | localhost | 57638
|
|
560160 | localhost | 57638
|
|
560160 | localhost | 57637
|
|
(10 rows)
|
|
|
|
DROP TABLE numbers_append;
|
|
-- Test copy failures against connection failures
|
|
-- create and switch to test user
|
|
CREATE USER test_user;
|
|
NOTICE: not propagating CREATE ROLE/USER commands to worker nodes
|
|
HINT: Connect to worker nodes directly to manually create all necessary users and roles.
|
|
SELECT * FROM run_command_on_workers('CREATE USER test_user');
|
|
nodename | nodeport | success | result
|
|
---------------------------------------------------------------------
|
|
localhost | 57637 | t | CREATE ROLE
|
|
localhost | 57638 | t | CREATE ROLE
|
|
(2 rows)
|
|
|
|
\c - test_user
|
|
SET citus.shard_count to 4;
|
|
CREATE TABLE numbers_hash (a int, b int);
|
|
SELECT create_distributed_table('numbers_hash', 'a', colocate_with:='none');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
COPY numbers_hash FROM STDIN WITH (FORMAT 'csv');
|
|
-- verify each placement is active
|
|
SELECT shardid, shardstate, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_hash'::regclass order by shardid, nodeport;
|
|
shardid | shardstate | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
560161 | 1 | localhost | 57637
|
|
560161 | 1 | localhost | 57638
|
|
560162 | 1 | localhost | 57637
|
|
560162 | 1 | localhost | 57638
|
|
560163 | 1 | localhost | 57637
|
|
560163 | 1 | localhost | 57638
|
|
560164 | 1 | localhost | 57637
|
|
560164 | 1 | localhost | 57638
|
|
(8 rows)
|
|
|
|
-- create a reference table
|
|
CREATE TABLE numbers_reference(a int, b int);
|
|
SELECT create_reference_table('numbers_reference');
|
|
create_reference_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
COPY numbers_reference FROM STDIN WITH (FORMAT 'csv');
|
|
-- create another hash distributed table
|
|
CREATE TABLE numbers_hash_other(a int, b int);
|
|
SELECT create_distributed_table('numbers_hash_other', 'a', colocate_with:='numbers_hash');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
SELECT shardid, shardstate, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_hash_other'::regclass order by shardid, nodeport;
|
|
shardid | shardstate | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
560166 | 1 | localhost | 57637
|
|
560166 | 1 | localhost | 57638
|
|
560167 | 1 | localhost | 57637
|
|
560167 | 1 | localhost | 57638
|
|
560168 | 1 | localhost | 57637
|
|
560168 | 1 | localhost | 57638
|
|
560169 | 1 | localhost | 57637
|
|
560169 | 1 | localhost | 57638
|
|
(8 rows)
|
|
|
|
-- manually corrupt pg_dist_shard such that both copies of one shard is placed in
|
|
-- worker_1+10. This is to test the behavior when no replica of a shard is accessible.
|
|
-- Whole copy operation is supposed to fail and rollback. We use session_replication_role
|
|
-- trick to disable the triggers on the pg_dist_shard_placement view
|
|
\c - :default_user
|
|
SET session_replication_role = replica;
|
|
UPDATE pg_dist_shard_placement SET nodeport = :worker_1_port+10 WHERE shardid = 560176 and nodeport = :worker_2_port;
|
|
SET session_replication_role = DEFAULT;
|
|
-- disable test_user on the first worker
|
|
\c - :default_user - :worker_1_port
|
|
SET citus.enable_metadata_sync TO off;
|
|
ALTER USER test_user WITH nologin;
|
|
\c - test_user - :master_port
|
|
-- reissue copy, and it should fail
|
|
COPY numbers_hash FROM STDIN WITH (FORMAT 'csv');
|
|
ERROR: connection to the remote node localhost:xxxxx failed with the following error: FATAL: role "test_user" is not permitted to log in
|
|
-- verify shards in the none of the workers as marked invalid
|
|
SELECT shardid, shardstate, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_hash'::regclass order by shardid, nodeport;
|
|
shardid | shardstate | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
560161 | 1 | localhost | 57637
|
|
560161 | 1 | localhost | 57638
|
|
560162 | 1 | localhost | 57637
|
|
560162 | 1 | localhost | 57638
|
|
560163 | 1 | localhost | 57637
|
|
560163 | 1 | localhost | 57638
|
|
560164 | 1 | localhost | 57637
|
|
560164 | 1 | localhost | 57638
|
|
(8 rows)
|
|
|
|
-- try to insert into a reference table copy should fail
|
|
COPY numbers_reference FROM STDIN WITH (FORMAT 'csv');
|
|
ERROR: connection to the remote node localhost:xxxxx failed with the following error: FATAL: role "test_user" is not permitted to log in
|
|
-- verify shards for reference table are still valid
|
|
SELECT shardid, shardstate, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_reference'::regclass order by placementid;
|
|
shardid | shardstate | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
560165 | 1 | localhost | 57637
|
|
560165 | 1 | localhost | 57638
|
|
(2 rows)
|
|
|
|
-- try to insert into numbers_hash_other. copy should fail and rollback
|
|
-- since it can not insert into either copies of a shard. shards are expected to
|
|
-- stay valid since the operation is rolled back.
|
|
COPY numbers_hash_other FROM STDIN WITH (FORMAT 'csv');
|
|
ERROR: connection to the remote node localhost:xxxxx failed with the following error: FATAL: role "test_user" is not permitted to log in
|
|
-- verify shards for numbers_hash_other are still valid
|
|
-- since copy has failed altogether
|
|
SELECT shardid, shardstate, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_hash_other'::regclass order by shardid, nodeport;
|
|
shardid | shardstate | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
560166 | 1 | localhost | 57637
|
|
560166 | 1 | localhost | 57638
|
|
560167 | 1 | localhost | 57637
|
|
560167 | 1 | localhost | 57638
|
|
560168 | 1 | localhost | 57637
|
|
560168 | 1 | localhost | 57638
|
|
560169 | 1 | localhost | 57637
|
|
560169 | 1 | localhost | 57638
|
|
(8 rows)
|
|
|
|
-- re-enable test_user on the first worker
|
|
\c - :default_user - :worker_1_port
|
|
SET citus.enable_metadata_sync TO off;
|
|
ALTER USER test_user WITH login;
|
|
\c - test_user - :master_port
|
|
DROP TABLE numbers_hash;
|
|
DROP TABLE numbers_hash_other;
|
|
DROP TABLE numbers_reference;
|
|
\c - :default_user
|
|
-- test copy failure inside the node
|
|
-- it will be done by changing definition of a shard table
|
|
SET citus.shard_count to 4;
|
|
SET citus.next_shard_id TO 560170;
|
|
CREATE TABLE numbers_hash(a int, b int);
|
|
SELECT create_distributed_table('numbers_hash', 'a', colocate_with:='none');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
\c - - - :worker_1_port
|
|
ALTER TABLE numbers_hash_560170 DROP COLUMN b;
|
|
\c - - - :master_port
|
|
-- operation will fail to modify a shard and roll back
|
|
COPY numbers_hash FROM STDIN WITH (FORMAT 'csv');
|
|
ERROR: column "b" of relation "numbers_hash_560170" does not exist
|
|
CONTEXT: while executing command on localhost:xxxxx
|
|
COPY numbers_hash, line 1: "1,1"
|
|
-- verify no row is inserted
|
|
SELECT count(a) FROM numbers_hash;
|
|
count
|
|
---------------------------------------------------------------------
|
|
0
|
|
(1 row)
|
|
|
|
-- verify shard is still marked as valid
|
|
SELECT shardid, shardstate, nodename, nodeport
|
|
FROM pg_dist_shard_placement join pg_dist_shard using(shardid)
|
|
WHERE logicalrelid = 'numbers_hash'::regclass order by shardid, nodeport;
|
|
shardid | shardstate | nodename | nodeport
|
|
---------------------------------------------------------------------
|
|
560170 | 1 | localhost | 57637
|
|
560170 | 1 | localhost | 57638
|
|
560171 | 1 | localhost | 57637
|
|
560171 | 1 | localhost | 57638
|
|
560172 | 1 | localhost | 57637
|
|
560172 | 1 | localhost | 57638
|
|
560173 | 1 | localhost | 57637
|
|
560173 | 1 | localhost | 57638
|
|
(8 rows)
|
|
|
|
DROP TABLE numbers_hash;
|
|
SELECT * FROM run_command_on_workers('DROP USER test_user');
|
|
nodename | nodeport | success | result
|
|
---------------------------------------------------------------------
|
|
localhost | 57637 | t | DROP ROLE
|
|
localhost | 57638 | t | DROP ROLE
|
|
(2 rows)
|
|
|
|
DROP USER test_user;
|
|
-- Test copy with built-in type without binary output function
|
|
CREATE TABLE test_binaryless_builtin (
|
|
col1 aclitem NOT NULL,
|
|
col2 character varying(255) NOT NULL
|
|
);
|
|
SELECT create_reference_table('test_binaryless_builtin');
|
|
create_reference_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
\COPY test_binaryless_builtin FROM STDIN WITH (format CSV)
|
|
SELECT * FROM test_binaryless_builtin;
|
|
col1 | col2
|
|
---------------------------------------------------------------------
|
|
postgres=r/postgres | test
|
|
(1 row)
|
|
|
|
DROP TABLE test_binaryless_builtin;
|
|
-- Test drop table with copy in the same transaction
|
|
BEGIN;
|
|
CREATE TABLE tt1(id int);
|
|
SELECT create_distributed_table('tt1','id');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
\copy tt1 from STDIN;
|
|
DROP TABLE tt1;
|
|
END;
|
|
-- Test dropping a column in front of the partition column
|
|
CREATE TABLE drop_copy_test_table (col1 int, col2 int, col3 int, col4 int);
|
|
SELECT create_distributed_table('drop_copy_test_table','col3');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
ALTER TABLE drop_copy_test_table drop column col1;
|
|
COPY drop_copy_test_table (col2,col3,col4) from STDIN with CSV;
|
|
SELECT * FROM drop_copy_test_table WHERE col3 = 1;
|
|
col2 | col3 | col4
|
|
---------------------------------------------------------------------
|
|
| 1 |
|
|
(1 row)
|
|
|
|
ALTER TABLE drop_copy_test_table drop column col4;
|
|
COPY drop_copy_test_table (col2,col3) from STDIN with CSV;
|
|
SELECT * FROM drop_copy_test_table WHERE col3 = 1;
|
|
col2 | col3
|
|
---------------------------------------------------------------------
|
|
| 1
|
|
| 1
|
|
(2 rows)
|
|
|
|
DROP TABLE drop_copy_test_table;
|
|
-- There should be no "tt1" shard on the worker nodes
|
|
\c - - - :worker_1_port
|
|
SELECT relname FROM pg_class WHERE relname LIKE 'tt1%';
|
|
relname
|
|
---------------------------------------------------------------------
|
|
(0 rows)
|
|
|
|
\c - - - :master_port
|
|
-- copy >8MB to a single worker to trigger a flush in PutRemoteCopyData
|
|
BEGIN;
|
|
CREATE UNLOGGED TABLE trigger_flush AS
|
|
SELECT 1 AS a, s AS b, s AS c, s AS d, s AS e, s AS f, s AS g, s AS h FROM generate_series(1,150000) s;
|
|
SELECT create_distributed_table('trigger_flush','a');
|
|
NOTICE: Copying data from local table...
|
|
NOTICE: copying the data has completed
|
|
DETAIL: The local data in the table is no longer visible, but is still on disk.
|
|
HINT: To remove the local data, run: SELECT truncate_local_data_after_distributing_table($$public.trigger_flush$$)
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
ABORT;
|
|
-- trigger switch-over when using single connection per worker
|
|
BEGIN;
|
|
SET citus.shard_count TO 3;
|
|
SET citus.multi_shard_modify_mode TO 'sequential';
|
|
CREATE UNLOGGED TABLE trigger_switchover(a int, b int, c int, d int, e int, f int, g int, h int);
|
|
SELECT create_distributed_table('trigger_switchover','a');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
INSERT INTO trigger_switchover
|
|
SELECT s AS a, s AS b, s AS c, s AS d, s AS e, s AS f, s AS g, s AS h FROM generate_series(1,250000) s;
|
|
ABORT;
|
|
-- copy into a table with a JSONB column
|
|
CREATE TABLE copy_jsonb (key text, value jsonb, extra jsonb default '["default"]'::jsonb);
|
|
SELECT create_distributed_table('copy_jsonb', 'key', colocate_with => 'none');
|
|
create_distributed_table
|
|
---------------------------------------------------------------------
|
|
|
|
(1 row)
|
|
|
|
-- JSONB from text should work
|
|
\COPY copy_jsonb (key, value) FROM STDIN
|
|
SELECT * FROM copy_jsonb ORDER BY key;
|
|
key | value | extra
|
|
---------------------------------------------------------------------
|
|
blue | {"b": 255, "g": 0, "r": 0} | ["default"]
|
|
green | {"b": 0, "g": 255, "r": 0} | ["default"]
|
|
(2 rows)
|
|
|
|
-- JSONB from binary should work
|
|
COPY copy_jsonb TO :'temp_dir''copy_jsonb.pgcopy' WITH (format binary);
|
|
COPY copy_jsonb FROM :'temp_dir''copy_jsonb.pgcopy' WITH (format binary);
|
|
SELECT * FROM copy_jsonb ORDER BY key;
|
|
key | value | extra
|
|
---------------------------------------------------------------------
|
|
blue | {"b": 255, "g": 0, "r": 0} | ["default"]
|
|
blue | {"b": 255, "g": 0, "r": 0} | ["default"]
|
|
green | {"b": 0, "g": 255, "r": 0} | ["default"]
|
|
green | {"b": 0, "g": 255, "r": 0} | ["default"]
|
|
(4 rows)
|
|
|
|
-- JSONB parsing error without validation: no line number
|
|
\COPY copy_jsonb (key, value) FROM STDIN
|
|
ERROR: invalid input syntax for json
|
|
DETAIL: The input string ended unexpectedly.
|
|
CONTEXT: JSON data, line 1: {"r":255,"g":0,"b":0
|
|
COPY copy_jsonb, line 1, column value: "{"r":255,"g":0,"b":0"
|
|
TRUNCATE copy_jsonb;
|
|
SET citus.skip_jsonb_validation_in_copy TO off;
|
|
-- JSONB from text should work
|
|
\COPY copy_jsonb (key, value) FROM STDIN
|
|
SELECT * FROM copy_jsonb ORDER BY key;
|
|
key | value | extra
|
|
---------------------------------------------------------------------
|
|
blue | {"b": 255, "g": 0, "r": 0} | ["default"]
|
|
green | {"b": 0, "g": 255, "r": 0} | ["default"]
|
|
(2 rows)
|
|
|
|
-- JSONB from binary should work
|
|
COPY copy_jsonb TO :'temp_dir''copy_jsonb.pgcopy' WITH (format binary);
|
|
COPY copy_jsonb FROM :'temp_dir''copy_jsonb.pgcopy' WITH (format binary);
|
|
SELECT * FROM copy_jsonb ORDER BY key;
|
|
key | value | extra
|
|
---------------------------------------------------------------------
|
|
blue | {"b": 255, "g": 0, "r": 0} | ["default"]
|
|
blue | {"b": 255, "g": 0, "r": 0} | ["default"]
|
|
green | {"b": 0, "g": 255, "r": 0} | ["default"]
|
|
green | {"b": 0, "g": 255, "r": 0} | ["default"]
|
|
(4 rows)
|
|
|
|
-- JSONB parsing error with validation: should see line number
|
|
\COPY copy_jsonb (key, value) FROM STDIN
|
|
ERROR: invalid input syntax for json
|
|
DETAIL: The input string ended unexpectedly.
|
|
CONTEXT: JSON data, line 1: {"r":255,"g":0,"b":0
|
|
COPY copy_jsonb, line 1, column value: "{"r":255,"g":0,"b":0"
|
|
DROP TABLE copy_jsonb;
|