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

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

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

-- 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 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

CREATE OR REPLACE FUNCTION public.drain_node(remove_host text, remove_port int DEFAULT 5432, shard_transfer_mode citus.shard_transfer_mode DEFAULT 'auto'::citus.shard_transfer_mode)
RETURNS void
LANGUAGE plpgsql
AS $function$
DECLARE
    coordinator_name text;
    coordinator_port int;
    candidate_count int;
    candidate_hosts text[];
    candidate_ports int[];

marcocitus

CREATE OR REPLACE FUNCTION public.repair_all()
RETURNS void
LANGUAGE plpgsql
AS $function$
BEGIN
PERFORM 
  master_copy_shard_placement(shardid, h.nodename, h.nodeport, u.nodename, u.nodeport)
FROM
  (SELECT shardid, nodename, nodeport FROM pg_dist_shard_placement WHERE shardstate = 3) u
JOIN

marcocitus

-- Create the raw events table
CREATE TABLE page_views (
    site_id int,
    path text,
    client_ip inet,
    view_time timestamptz default now(),
    view_id bigserial
);

-- Allow fast lookups of ranges of sequence IDs

marcocitus

/* add the user mappings from the distributed table to the shards */
CREATE OR REPLACE FUNCTION bind_shards_to_remote_shards(table_name regclass)
RETURNS void
LANGUAGE plpgsql
AS $function$
DECLARE
    foreign_server text;
    fdw_options text[];
    user_mapping_options text[];
    server_options text[];

marcocitus

#!/bin/sh
name=Marco
formation_name=${1:-test}

formations_url=https://console.citusdata.com/api/v1/formations
api_token=$(cat $HOME/.cituscloud/token)
formation_dir=$HOME/.cituscloud/formations/$formation_name/
worker_count=2

function log {