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

create table citus_schema_isolation (
    schemaid regnamespace,
    nodeid int,
    primary key (schemaid)
);

CREATE OR REPLACE FUNCTION shard_allowed_on_node_with_schema_isolation(p_shardid bigint, p_nodeid integer)
 RETURNS boolean
 LANGUAGE plpgsql
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 TABLE IF NOT EXISTS words (
  word text
);

CREATE TABLE IF NOT EXISTS product (
  product_id int not null, 
  name text not null,
  description text not null,
  price decimal(12,2),
  attributes jsonb,

marcocitus

-- set up a global cluster
select citus_set_coordinator_host('us-coordinator-hostname', 5432);
select citus_add_node('us-worker1-hostname', 5432);
select citus_add_node('us-worker2-hostname', 5432);
select citus_add_node('eu-coordinator-hostname', 5432);
select citus_add_node('eu-worker1-hostname', 5432);
select citus_add_node('eu-worker2-hostname', 5432);

-- make sure we can run distributed queries on EU coordinator
select start_metadata_sync_to_node('eu-coordinator-hostname', 5432);

marcocitus

DROP TABLE IF EXISTS data, data_writes;

CREATE TABLE data (
    key int primary key,
    value int
);
SELECT create_distributed_table('data','key');

CREATE TABLE data_writes (
    LIKE data

marcocitus

CREATE OR REPLACE FUNCTION latency(nodename text, nodeport int)
RETURNS double precision
AS $$
DECLARE
    start_time timestamptz;
    end_time timestamptz;
BEGIN
    /* warm up connection cache */
    perform master_run_on_worker(array[nodename], array[nodeport], array['select 1'], false);

marcocitus

-- PG set up instructions: https://ankane.org/tpc-ds

SELECT create_reference_table('call_center');
SELECT create_reference_table('catalog_page');
SELECT create_distributed_table('catalog_returns', 'cr_item_sk');
SELECT create_distributed_table('catalog_sales', 'cs_item_sk');
SELECT create_reference_table('customer');
SELECT create_reference_table('customer_address');
SELECT create_reference_table('customer_demographics');
SELECT create_reference_table('date_dim');

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 alter_table_access_method(table_name regclass, access_method name)
RETURNS void
LANGUAGE plpgsql 
AS $$
DECLARE
        compressed_table_name name := table_name || '_cmp';
        original_table_name name := table_name::name;
        table_relkind char; 
        parent_table_name regclass;
        partition_boundaries text;