marcocitus

CREATE TABLE rollups (
  name text,
  rolled_up_generation bigint default -1
);

-- Create a stub on workers to allow usage as a default in distributed tables
SELECT run_command_on_workers($$
CREATE OR REPLACE FUNCTION current_rollup_generation(rollup_name text)
RETURNS bigint LANGUAGE sql
AS $function$

marcocitus

-- as superuser:

CREATE EXTENSION pg_cron; 
GRANT USAGE ON SCHEMA cron TO citus;

CREATE SCHEMA partman;
GRANT USAGE ON SCHEMA partman TO citus;
CREATE EXTENSION pg_partman WITH SCHEMA partman;
GRANT ALL ON TABLE partman.part_config TO citus; 
GRANT ALL ON TABLE partman.part_config_sub TO citus;

marcocitus

CREATE OR REPLACE FUNCTION public.load_github_data(events_date date, hour integer)
 RETURNS void
 LANGUAGE plpgsql
 SECURITY DEFINER
AS $function$
BEGIN
    CREATE TEMPORARY TABLE input (data jsonb) ON COMMIT DROP;
    EXECUTE format($$COPY input FROM PROGRAM 'curl -s http://data.githubarchive.org/%s-%s.json.gz | zcat | grep -v \\u0000' CSV QUOTE e'\x01' DELIMITER e'\x02'$$, events_date, hour);
    INSERT INTO github.events SELECT
        (data->>'id')::bigint AS event_id,

marcocitus

CREATE OR REPLACE FUNCTION create_colocated_range_shards(table_name regclass, min_value text, max_value text)
RETURNS void
LANGUAGE plpgsql
AS $function$
DECLARE
    table_colocationid int;
    group_ids int[];
    new_shard_id bigint;
    current_table text;
    node record;

marcocitus

/* commands to run on the coordinator */

CREATE EXTENSION citus;
SELECT master_add_node('10.0.0.2', 5432);
SELECT master_add_node('10.0.0.3', 5432);
SELECT start_metadata_sync_to_node(nodename, nodeport) FROM pg_dist_node;
SET citus.replication_model TO 'streaming'

CREATE TABLE events (
  event_id bigserial primary key,

marcocitus

-- Replace the pg_table_size function with one that supports distributed tables
BEGIN;
ALTER FUNCTION pg_catalog.pg_table_size(regclass) RENAME TO orig_table_size;

CREATE OR REPLACE FUNCTION pg_catalog.pg_table_size(table_name regclass)
RETURNS bigint
LANGUAGE plpgsql
AS $function$
BEGIN
        IF EXISTS (SELECT 1 FROM pg_dist_partition WHERE logicalrelid = table_name) THEN

marcocitus

CREATE OR REPLACE FUNCTION run_command_on_master_and_workers(p_sql text)
RETURNS void LANGUAGE plpgsql AS $function$
BEGIN
     EXECUTE p_sql;
     PERFORM run_command_on_workers(p_sql);
END;
$function$;

SELECT run_command_on_master_and_workers($cmd$
CREATE SCHEMA IF NOT EXISTS citus_split_lower

marcocitus

CREATE TABLE events (
  id int not null,
  created_at timestamptz,
  payload jsonb
);
CREATE INDEX ON events (id);
CREATE INDEX ON events USING BRIN (created_at);
SELECT create_distributed_table('events', 'id');

-- Set up the trigger on workers (idempotent)

marcocitus

BEGIN;

DROP FUNCTION public.master_append_table_to_shard(bigint,text,text,int);

ALTER FUNCTION pg_catalog.real_master_append_table_to_shard(bigint,text,text,int)
RENAME TO master_append_table_to_shard;

END;

marcocitus

CREATE OR REPLACE FUNCTION master_create_range_shard(
                                                        table_name text,
                                                        minvalue text, 
                                                        maxvalue text)
RETURNS void
LANGUAGE plpgsql
AS $function$
DECLARE
        new_shard_id bigint;
BEGIN