-- ===================================================================
-- create test functions
-- ===================================================================
CREATE FUNCTION generate_alter_table_detach_partition_command(regclass)
    RETURNS text
    AS 'citus'
    LANGUAGE C STRICT;

CREATE FUNCTION generate_alter_table_attach_partition_command(regclass)
    RETURNS text
    AS 'citus'
    LANGUAGE C STRICT;

CREATE FUNCTION generate_partition_information(regclass)
    RETURNS text
    AS 'citus'
    LANGUAGE C STRICT;

CREATE FUNCTION print_partitions(regclass)
    RETURNS text
    AS 'citus'
    LANGUAGE C STRICT;

CREATE FUNCTION table_inherits(regclass)
    RETURNS bool
    AS 'citus'
    LANGUAGE C STRICT;

CREATE FUNCTION table_inherited(regclass)
    RETURNS bool
    AS 'citus'
    LANGUAGE C STRICT;

CREATE OR REPLACE FUNCTION detach_and_attach_partition(partition_name regclass, parent_table_name regclass)
RETURNS void LANGUAGE plpgsql VOLATILE
AS $function$
DECLARE
  detach_partition_command text := '';
	attach_partition_command text := '';
	command_result text := '';

BEGIN
   -- first generate the command
   SELECT public.generate_alter_table_attach_partition_command(partition_name) INTO attach_partition_command;

   -- now genereate the detach command
   SELECT public.generate_alter_table_detach_partition_command(partition_name) INTO detach_partition_command;

   -- later detach the same partition
   EXECUTE detach_partition_command;

   -- not attach it again
   EXECUTE attach_partition_command;
END;
$function$;

CREATE OR REPLACE FUNCTION drop_and_recreate_partitioned_table(parent_table_name regclass)
RETURNS void LANGUAGE plpgsql VOLATILE
AS $function$
DECLARE
	command text := '';

BEGIN
   -- first generate the command
   CREATE TABLE partitioned_table_create_commands AS SELECT master_get_table_ddl_events(parent_table_name::text);

   -- later detach the same partition
   EXECUTE 'DROP TABLE ' || parent_table_name::text || ';';

   FOR command IN SELECT * FROM partitioned_table_create_commands
   LOOP
        -- can do some processing here
   		EXECUTE command;
    END LOOP;

    DROP TABLE partitioned_table_create_commands;

END;
$function$;

-- create a partitioned table
CREATE TABLE date_partitioned_table(id int, time date) PARTITION BY RANGE (time);

-- we should be able to get the partitioning information even if there are no partitions
SELECT generate_partition_information('date_partitioned_table');

-- we should be able to drop and re-create the partitioned table using the command that Citus generate
SELECT drop_and_recreate_partitioned_table('date_partitioned_table');

-- we should also be able to see the PARTITION BY ... for the parent table
SELECT master_get_table_ddl_events('date_partitioned_table');

-- now create the partitions
CREATE TABLE date_partition_2006 PARTITION OF date_partitioned_table FOR VALUES FROM ('2006-01-01') TO ('2007-01-01');
CREATE TABLE date_partition_2007 PARTITION OF date_partitioned_table FOR VALUES FROM ('2007-01-01') TO ('2008-01-01');

-- we should be able to get the partitioning information after the partitions are created
SELECT generate_partition_information('date_partitioned_table');

-- lets get the attach partition commands
SELECT generate_alter_table_attach_partition_command('date_partition_2006');
SELECT generate_alter_table_attach_partition_command('date_partition_2007');

-- detach and attach the partition by the command generated by us
\d+ date_partitioned_table

SELECT detach_and_attach_partition('date_partition_2007', 'date_partitioned_table');

-- check that both partitions are visiable
\d+ date_partitioned_table


-- make sure that inter shard commands work as expected
-- assume that the shardId is 100
CREATE TABLE date_partitioned_table_100 (id int, time date) PARTITION BY RANGE (time);
CREATE TABLE date_partition_2007_100 (id int, time date );

-- now create the partitioning hierarcy
SELECT worker_apply_inter_shard_ddl_command(referencing_shard:=100, referencing_schema_name:='public',
											referenced_shard:=100, referenced_schema_name:='public',
											command:='ALTER TABLE date_partitioned_table ATTACH PARTITION date_partition_2007 FOR VALUES FROM (''2007-01-01'') TO (''2008-01-02'')' );

-- the hierarcy is successfully created
\d+ date_partitioned_table_100

-- Citus can also get the DDL events for the partitions as regular tables
SELECT master_get_table_ddl_events('date_partition_2007_100');

-- now break the partitioning hierarcy
SELECT worker_apply_inter_shard_ddl_command(referencing_shard:=100, referencing_schema_name:='public',
                      referenced_shard:=100, referenced_schema_name:='public',
                      command:='ALTER TABLE date_partitioned_table DETACH PARTITION date_partition_2007' );

-- the hierarcy is successfully broken
\d+ date_partitioned_table_100

-- now lets have some more complex partitioning hierarcies with
-- tables on different schemas and constraints on the tables

CREATE SCHEMA partition_parent_schema;
CREATE TABLE partition_parent_schema.parent_table (id int NOT NULL, time date DEFAULT now()) PARTITION BY RANGE (time);

CREATE SCHEMA partition_child_1_schema;
CREATE TABLE partition_child_1_schema.child_1 (id int NOT NULL, time date );

CREATE SCHEMA partition_child_2_schema;
CREATE TABLE partition_child_2_schema.child_2 (id int NOT NULL, time date );

-- we should be able to get the partitioning information even if there are no partitions
SELECT generate_partition_information('partition_parent_schema.parent_table');

-- we should be able to drop and re-create the partitioned table using the command that Citus generate
SELECT drop_and_recreate_partitioned_table('partition_parent_schema.parent_table');

ALTER TABLE partition_parent_schema.parent_table ATTACH PARTITION partition_child_1_schema.child_1 FOR VALUES FROM ('2009-01-01') TO ('2010-01-02');


SET search_path = 'partition_parent_schema';

ALTER TABLE parent_table ATTACH PARTITION  partition_child_2_schema.child_2 FOR VALUES FROM  ('2006-01-01') TO ('2007-01-01');

SELECT public.generate_partition_information('parent_table');

-- lets get the attach partition commands
SELECT public.generate_alter_table_attach_partition_command('partition_child_1_schema.child_1');

SET search_path = 'partition_child_2_schema';

SELECT public.generate_alter_table_attach_partition_command('child_2');

SET search_path = 'partition_parent_schema';

-- detach and attach the partition by the command generated by us
\d+ parent_table

SELECT public.detach_and_attach_partition('partition_child_1_schema.child_1', 'parent_table');

-- check that both partitions are visiable
\d+ parent_table

-- some very simple checks that should error out
SELECT public.generate_alter_table_attach_partition_command('parent_table');
SELECT public.generate_partition_information('partition_child_1_schema.child_1');
SELECT public.print_partitions('partition_child_1_schema.child_1');

-- now pring the partitions
SELECT public.print_partitions('parent_table');

SET search_path = 'public';

-- test multi column / expression partitioning with UNBOUNDED ranges
CREATE OR REPLACE FUNCTION some_function(input_val text)
RETURNS text LANGUAGE plpgsql IMMUTABLE
AS $function$
BEGIN
		return reverse(input_val);
END;
$function$;

CREATE TABLE multi_column_partitioned (
   a int,
   b int,
   c text
 ) PARTITION BY RANGE (a, (a+b+1), some_function(upper(c)));

CREATE TABLE multi_column_partition_1(
   a int,
   b int,
   c text
);

CREATE TABLE multi_column_partition_2(
   a int,
   b int,
   c text
);

-- partitioning information
SELECT generate_partition_information('multi_column_partitioned');
SELECT master_get_table_ddl_events('multi_column_partitioned');
SELECT drop_and_recreate_partitioned_table('multi_column_partitioned');

-- partitions and their ranges
ALTER TABLE multi_column_partitioned ATTACH PARTITION multi_column_partition_1 FOR VALUES FROM (1, 10, '250') TO (1, 20, '250');
SELECT generate_alter_table_attach_partition_command('multi_column_partition_1');
ALTER TABLE multi_column_partitioned ATTACH PARTITION multi_column_partition_2 FOR VALUES FROM (10, 1000, '2500') TO (MAXVALUE, MAXVALUE, MAXVALUE);
SELECT generate_alter_table_attach_partition_command('multi_column_partition_2');
SELECT generate_alter_table_detach_partition_command('multi_column_partition_2');

-- finally a test with LIST partitioning
CREATE TABLE list_partitioned (col1 NUMERIC, col2 NUMERIC, col3 VARCHAR(10)) PARTITION BY LIST (col1) ;

SELECT generate_partition_information('list_partitioned');
SELECT master_get_table_ddl_events('list_partitioned');
SELECT drop_and_recreate_partitioned_table('list_partitioned');

CREATE TABLE list_partitioned_1 PARTITION OF list_partitioned FOR VALUES IN (100, 101, 102, 103, 104);

SELECT generate_alter_table_attach_partition_command('list_partitioned_1');

-- also differentiate partitions and inhereted tables
CREATE TABLE cities (
    name            text,
    population      float,
    altitude        int     -- in feet
);

CREATE TABLE capitals (
    state           char(2)
) INHERITS (cities);

-- returns true since capitals inherits from cities
SELECT table_inherits('capitals');

-- although date_partition_2006 inherits from its parent
-- returns false since the hierarcy is formed via partitioning
SELECT table_inherits('date_partition_2006');

-- returns true since cities inherited by capitals
SELECT table_inherited('cities');

-- although date_partitioned_table inherited by its partitions
-- returns false since the hierarcy is formed via partitioning
SELECT table_inherited('date_partitioned_table');

-- also these are not supported
SELECT master_get_table_ddl_events('capitals');
SELECT master_get_table_ddl_events('cities');

-- dropping parents frop the partitions
DROP TABLE date_partitioned_table, multi_column_partitioned, list_partitioned, partition_parent_schema.parent_table, cities, capitals;