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

marcocitus

-- Functions to partition a table and drop old partitions

CREATE SEQUENCE IF NOT EXISTS partition_id_seq;


CREATE OR REPLACE FUNCTION partition_create(
                             table_name regclass)
RETURNS text
LANGUAGE plpgsql
AS $function$

marcocitus

CREATE OR REPLACE FUNCTION citus_shard_name(table_name regclass, shard_id bigint)
RETURNS text
LANGUAGE sql
AS $function$
  SELECT table_name||'_'||shard_id;
$function$;

CREATE OR REPLACE FUNCTION citus_shard_name(shard_id bigint)
RETURNS text
LANGUAGE sql

marcocitus

CREATE OR REPLACE FUNCTION consecutive_date_pattern(start_date date, num_days int)
RETURNS text[]
AS $BODY$
DECLARE
  result text[];
BEGIN
  SELECT array_agg('date=>"'||d::date||'"') INTO result
  FROM generate_series(start_date, start_date + num_days * interval '1 day', '1 day') d;

  RETURN result;

marcocitus

--
-- Test compatibility with Citus extension
--

-- Function to perform a COPY, but hide error messages
CREATE OR REPLACE FUNCTION silent_copy()
RETURNS void LANGUAGE plpgsql AS $function$
BEGIN
    COPY contestant FROM '@abs_srcdir@/data/contestants.1.csv' WITH CSV;
EXCEPTION WHEN OTHERS THEN

marcocitus

postgres=# EXPLAIN ANALYZE SELECT p.product_id, p.name, p.price, to_json(array_agg(to_json(o)))
FROM   product p LEFT JOIN offer o USING (product_id)
WHERE  product_text_search(p.name, p.description) @@ plainto_tsquery('copper oven')
  AND (p.price < 70 OR o.price < 70)
  AND (p.attributes @> '{"food":"waste"}' OR p.attributes @> '{"food":"air"}')
GROUP BY p.product_id, p.name, p.description, p.price
ORDER BY ts_rank(product_text_search(p.name, p.description),
                 plainto_tsquery('copper oven')) DESC
LIMIT 10;
                                                                                                             QUERY PLAN