bucardo.schema

-- Schema for the main Bucardo database

-- Dependencies:
-- Superuser bucardo owns a database named bucardo

-- plpgsql and plperlu are loaded, but just in case:
SET client_min_messages = 'FATAL';
CREATE LANGUAGE plpgsql;
CREATE LANGUAGE plperlu;
SET client_min_messages = 'ERROR';
COMMIT;

\set ON_ERROR_STOP

CREATE SCHEMA bucardo;
CREATE SCHEMA freezer;
SET search_path TO bucardo;

SET client_min_messages   = 'WARNING';
SET escape_string_warning = 'OFF';

--
-- Main bucardo configuration information
--
CREATE TABLE bucardo_config (
  setting  TEXT        NOT NULL, -- short unique name, maps to %config inside Bucardo
  value    TEXT        NOT NULL,
  about    TEXT            NULL, -- long description
  type     TEXT            NULL, -- sync or goat
  name     TEXT            NULL, -- which specific sync or goat
  cdate    TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE UNIQUE INDEX bucardo_config_unique ON bucardo_config(setting) WHERE name IS NULL;

CREATE UNIQUE INDEX bucardo_config_unique_name ON bucardo_config(setting,name,type) WHERE name IS NOT NULL;

ALTER TABLE bucardo_config ADD CONSTRAINT valid_config_type
  CHECK (type IN ('sync','goat'));

CREATE FUNCTION check_bucardo_config()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $bc$
BEGIN
NEW.setting = LOWER(NEW.setting);

IF (NEW.type IS NOT NULL and NEW.name IS NULL) THEN
  RAISE EXCEPTION 'Must provide a specific %', NEW.type;
END IF;

IF (NEW.name IS NOT NULL and NEW.type IS NULL) THEN
  RAISE EXCEPTION 'Must provide a type if giving a name';
END IF;

IF (NEW.setting = 'sync' OR NEW.setting = 'goat') THEN
  RAISE EXCEPTION 'Invalid setting name';
END IF;

RETURN NEW;
END;
$bc$;

CREATE TRIGGER check_bucardo_config
  BEFORE INSERT OR UPDATE ON bucardo_config
  FOR EACH ROW EXECUTE PROCEDURE check_bucardo_config();


-- Sleep times (all in seconds)
COPY bucardo_config(setting,value,about)
FROM STDIN
WITH DELIMITER '|';
kick_sleep|0.2|How long do we sleep while waiting for a kick response?
mcp_loop_sleep|0.1|How long does the main MCP daemon sleep between loops?
mcp_dbproblem_sleep|15|How many seconds to sleep before trying to respawn
ctl_nothingfound_sleep|0.2|How long does the controller loop sleep if nothing is found?
kid_nothingfound_sleep|0.3|How long does a kid sleep if nothing is found?
kid_nodeltarows_sleep|0.8|How long do kids sleep if no delta rows are found?
kid_serial_sleep|10|How long to sleep in seconds if we hit a serialization error
endsync_sleep|1.0|How long do we sleep when custom code requests an endsync?
\.


-- Various timeouts (times are in seconds)
COPY bucardo_config(setting,value,about)
FROM STDIN
WITH DELIMITER '|';
mcp_pingtime|60|How often do we ping check the MCP?
ctl_pingtime|600|How often do we ping check the CTL?
kid_pingtime|60|How often do we ping check the KID?
ctl_checkonkids_time|10|How often does the controller check on the kids health?
ctl_checkabortedkids_time|30|How often does the controller check the q table for aborted children?
ctl_createkid_time|0.5|How long do we sleep to allow kids-on-demand to get on their feet?
tcp_keepalives_idle|10|How long to wait between each keepalive probe.
tcp_keepalives_interval|5|How long to wait for a response to a keepalive probe.
tcp_keepalives_count|2|How many probes to send. 0 indicates sticking with system defaults.
\.

-- Debug output
COPY bucardo_config(setting,value,about)
FROM STDIN
WITH DELIMITER '|';
audit_pid|1|Do we populate the audit_pid table or not?
log_showpid|0|Show PID in the log output?
log_showtime|1|Show timestamp in the log output?  0=off  1=seconds since epoch  2=scalar gmtime  3=scalar localtime
log_showline|0|Show line number in the log output?
log_conflict_details|0|Log detailed conflict data?
log_conflict_file|bucardo_conflict.log|Name of the conflict detail log file
\.

-- Versioning
COPY bucardo_config(setting,value,about)
FROM STDIN
WITH DELIMITER '|';
bucardo_version|3.2.7|Bucardo version this schema was created with
bucardo_current_version|3.2.7|Current version of Bucardo
\.

-- Other settings:
COPY bucardo_config(setting,value,about)
FROM STDIN
WITH DELIMITER '|';
default_email_from|nobody@example.com|Who the alert emails are sent as
default_email_to|nobody@example.com|Who to send alert emails to
kid_abort_limit|3|How many times we will restore an aborted kid before giving up?
max_delete_clause|200|Maximum number of items to delete inside of IN() clauses
max_select_clause|500|Maximum number of items to select inside of IN() clauses
piddir|/var/run/bucardo|Directory holding Bucardo PID files
pidfile|bucardo.pid|Name of the main Bucardo pid file
reason_file|/home/bucardo/restart.reason|File to hold reasons for stopping and starting
stats_script_url|http://www.bucardo.org/|Location of the stats script
stopfile|fullstopbucardo|Name of the semaphore file used to stop Bucardo processes
syslog_facility|LOG_LOCAL1|Which syslog facility level to use
upsert_attempts|3|How many times do we try out the upsert loop?
\.

--
-- Keep track of every database we need to connect to
--
CREATE TABLE db (
  name        TEXT        NOT NULL,  -- local name for convenience, not necessarily database name
                CONSTRAINT db_name_pk PRIMARY KEY (name),
  dbhost      TEXT        NOT NULL DEFAULT '',
  dbport      TEXT        NOT NULL DEFAULT 5432,
  dbname      TEXT        NOT NULL,
  dbuser      TEXT        NOT NULL,
  dbpass      TEXT            NULL,
  dbconn      TEXT        NOT NULL DEFAULT '',  -- string to add to the generated dsn
  dbservice   TEXT            NULL,
  pgpass      TEXT            NULL,      -- local file with connection info same as pgpass
  status      TEXT        NOT NULL DEFAULT 'active',
  sourcelimit SMALLINT    NOT NULL DEFAULT 0, -- maximum concurrent read connections to this database
  targetlimit SMALLINT    NOT NULL DEFAULT 0, -- maximum concurrent write connections to this database
  cdate       TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE UNIQUE INDEX "db_dsn_unique" ON db(dbhost,dbport,dbname,dbuser) WHERE NOT name ~ '^bctest';
ALTER TABLE db ADD CONSTRAINT db_status CHECK (status IN ('active','inactive'));

--
-- Databases can belong to zero or more named groups
--
CREATE TABLE dbgroup (
  name      TEXT        NOT NULL,
              CONSTRAINT dbgroup_name_pk PRIMARY KEY (name),
  cdate     TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE dbmap (
  db       TEXT        NOT NULL,
             CONSTRAINT  dbmap_db_fk FOREIGN KEY (db) REFERENCES db(name) ON DELETE CASCADE,
  dbgroup  TEXT        NOT NULL,
             CONSTRAINT  dbmap_dbgroup_fk FOREIGN KEY (dbgroup) REFERENCES dbgroup(name) ON DELETE CASCADE,
  priority SMALLINT    NOT NULL DEFAULT 0,
  cdate    TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE UNIQUE INDEX dbmap_unique ON dbmap(db,dbgroup);


--
-- Track status information about each database
--
CREATE TABLE db_connlog (
  db          TEXT        NOT NULL,
                CONSTRAINT  db_connlog_dbid_fk FOREIGN KEY (db) REFERENCES db(name) ON DELETE CASCADE,
  conndate    TIMESTAMPTZ NOT NULL DEFAULT now(),  -- when we first connected to it
  connstring  TEXT        NOT NULL,
  status      TEXT        NOT NULL DEFAULT 'unknown',
                CONSTRAINT db_connlog_status CHECK (status IN ('unknown', 'good', 'down', 'unreachable')),
  version     TEXT            NULL
);


--
-- We need to track each item we want to replicate from or replicate to
--
CREATE SEQUENCE goat_id_seq;
CREATE TABLE goat (
  id                 INTEGER     NOT NULL DEFAULT nextval('goat_id_seq'),
                       CONSTRAINT goat_id_pk PRIMARY KEY (id),
  db                 TEXT        NOT NULL,
                       CONSTRAINT goat_db_fk FOREIGN KEY (db) REFERENCES db(name) ON DELETE RESTRICT,
  schemaname         TEXT        NOT NULL,
  tablename          TEXT        NOT NULL,
  reltype            TEXT        NOT NULL DEFAULT 'table',
  pkey               TEXT            NULL,
  qpkey              TEXT            NULL,
  pkeytype           TEXT            NULL,
  has_delta          BOOLEAN     NOT NULL DEFAULT 'false',
  ping               BOOLEAN         NULL,                 -- overrides sync-level ping
  customselect       TEXT            NULL,
  makedelta          BOOLEAN         NULL,
  rebuild_index      BOOLEAN         NULL,                 -- overrides sync-level rebuild_index
  ghost              BOOLEAN     NOT NULL DEFAULT 'false', -- only drop triggers, do not replicate
  standard_conflict  TEXT            NULL,
  analyze_after_copy BOOLEAN     NOT NULL DEFAULT 'true',
  strict_checking    BOOLEAN     NOT NULL DEFAULT 'true',
  cdate              TIMESTAMPTZ NOT NULL DEFAULT now()
);
ALTER TABLE goat ADD CONSTRAINT has_schemaname CHECK (length(schemaname) > 1);

ALTER TABLE goat ADD CONSTRAINT custom_needs_pkey CHECK (customselect IS NULL OR length(pkey) > 1);

ALTER TABLE goat ADD CONSTRAINT pkey_needs_type CHECK (pkey = '' OR pkeytype IS NOT NULL);

ALTER TABLE goat ADD CONSTRAINT standard_conflict CHECK (standard_conflict IS NULL
  OR standard_conflict IN ('source','target','skip','random','latest','abort'));

-- CREATE UNIQUE INDEX goats_r_unique ON goat(db,schemaname,tablename,reltype);

--
-- A group of goats. Ideally arranged in some sort of tree.
--
CREATE TABLE herd (
  name       TEXT        NOT NULL,
               CONSTRAINT herd_name_pk PRIMARY KEY (name),
  cdate      TIMESTAMPTZ NOT NULL DEFAULT now()
);

--
-- Goats belong to zero or more herds. In most cases, they will 
-- belong to a single her if they are being replicated.
--
CREATE TABLE herdmap (
  herd      TEXT        NOT NULL,
              CONSTRAINT herdmap_herd_fk FOREIGN KEY (herd) REFERENCES herd(name) ON DELETE CASCADE,
  goat      INTEGER     NOT NULL,
              CONSTRAINT herdmap_goat_fk FOREIGN KEY (goat) REFERENCES goat(id) ON DELETE CASCADE,
  priority  SMALLINT    NOT NULL DEFAULT 0,
  cdate     TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE VIEW goats_in_herd AS
  SELECT hm.herd, g.*
  FROM goat g
  JOIN herdmap hm ON hm.goat = g.id;

CREATE FUNCTION herdcheck()
RETURNS TRIGGER
LANGUAGE plperlu
AS
$bc$
    use strict; use warnings;
    ## Make sure that all goats in a herd are from the same database
    my $new = $_TD->{new};
    my $herdname = $new->{herd};
    $herdname =~ s/'/''/go;
    my $SQL = qq{
        SELECT 1
        FROM   bucardo.herdmap h, bucardo.goat g
        WHERE  h.goat = g.id
        AND    h.herd='$herdname'
        AND    g.db != (SELECT db FROM bucardo.goat WHERE id = $new->{goat})
    };
    elog(DEBUG, "Running $SQL");
    my $count = spi_exec_query($SQL)->{processed};
    if ($count >= 1) {
        elog(ERROR, "Cannot have goats from different databases in the same herd ($count)");
    }

    ## Make sure that a herd contains at most one schemaname/tablename combination
    $SQL = qq{
        SELECT count(*) AS goats
        FROM   bucardo.herdmap h, bucardo.goat g
        WHERE  h.goat = g.id
        AND    g.id != $new->{goat}
        AND    h.herd = '$herdname'
        AND    g.tablename = (SELECT tablename FROM bucardo.goat WHERE id = $new->{goat})
        AND    g.schemaname = (SELECT schemaname FROM bucardo.goat WHERE id = $new->{goat})
    };
    elog(DEBUG, "Running $SQL");
    $count = spi_exec_query($SQL)->{rows}[0]{goats};
    if ($count >= 1) {
        elog(ERROR, "Cannot have two goats with the same schema and table inside a herd (herd=$herdname) (goat=$new->{goat}) (count=$count)");
    }

    return;
$bc$;

CREATE TRIGGER herdcheck
  AFTER INSERT OR UPDATE ON herdmap
  FOR EACH ROW EXECUTE PROCEDURE herdcheck();


--
-- We need to know who is replicating to who, and how
--
CREATE TABLE sync (
  name               TEXT        NOT NULL UNIQUE,
                       CONSTRAINT sync_name_pk PRIMARY KEY (name),
  source             TEXT        NOT NULL,
                       CONSTRAINT sync_source_herd_fk FOREIGN KEY (source) REFERENCES herd(name) ON DELETE RESTRICT,
  targetdb           TEXT            NULL,
                       CONSTRAINT sync_targetdb_fk FOREIGN KEY (targetdb) REFERENCES db(name) ON DELETE RESTRICT,
  targetgroup        TEXT            NULL,
                       CONSTRAINT sync_targetgroup_fk FOREIGN KEY (targetgroup) REFERENCES dbgroup(name) ON DELETE RESTRICT,
  synctype           TEXT        NOT NULL,
  stayalive          BOOLEAN     NOT NULL DEFAULT 'true', -- Does the sync controller stay connected?
  kidsalive          BOOLEAN     NOT NULL DEFAULT 'true', -- Do the children stay connected?
  usecustomselect    BOOLEAN     NOT NULL DEFAULT 'false',
  copytype           TEXT        NOT NULL DEFAULT 'copy',
  copyextra          TEXT        NOT NULL DEFAULT '',  -- e.g. WITH OIDS
  deletemethod       TEXT        NOT NULL DEFAULT 'delete',
  limitdbs           SMALLINT    NOT NULL DEFAULT 0,     -- How many databases can sync at once? 0=all
  ping               BOOLEAN     NOT NULL DEFAULT true,      -- Are we issuing NOTICES via triggers?
  do_listen          BOOLEAN     NOT NULL DEFAULT false,     -- LISTEN for kicks on source/target database if ! ping
  checktime          INTERVAL        NULL,                   -- How often to check if we've not heard anything?
  status             TEXT        NOT NULL DEFAULT 'active',  -- Possibly CHECK / FK ('stopped','paused','b0rken')
  makedelta          BOOLEAN     NOT NULL DEFAULT 'false',
  rebuild_index      BOOLEAN     NOT NULL DEFAULT 'false',   -- Load without indexes and then REINDEX table
  priority           SMALLINT    NOT NULL DEFAULT 0,     -- Higher is better
  txnmode            TEXT        NOT NULL DEFAULT 'SERIALIZABLE',
  analyze_after_copy BOOLEAN     NOT NULL DEFAULT 'true',
  strict_checking    BOOLEAN     NOT NULL DEFAULT 'true',
  overdue            INTERVAL    NOT NULL DEFAULT '0 seconds'::interval,
  expired            INTERVAL    NOT NULL DEFAULT '0 seconds'::interval,
  track_rates        BOOLEAN     NOT NULL DEFAULT 'false',
  cdate              TIMESTAMPTZ NOT NULL DEFAULT now()
);


ALTER TABLE sync ADD CONSTRAINT sync_type CHECK (synctype IN ('pushdelta','fullcopy','swap'));

ALTER TABLE sync ADD CONSTRAINT sync_copytype CHECK (copytype IN ('insert','copy'));

ALTER TABLE sync ADD CONSTRAINT sync_deletemethod CHECK (deletemethod IN ('truncate', 'delete'));

CREATE UNIQUE INDEX sync_source_targetdb_type ON sync(source, targetdb, synctype);

CREATE UNIQUE INDEX sync_source_targetgroup_type ON sync(source, targetgroup, synctype);

ALTER TABLE sync ADD CONSTRAINT sync_validtarget CHECK
  ((targetdb IS NULL AND targetgroup IS NOT NULL)
    OR
   (targetdb IS NOT NULL AND targetgroup IS NULL));

ALTER TABLE sync ADD CONSTRAINT sync_swap_nogroup CHECK
  (synctype <> 'swap' or targetdb IS NOT NULL);

-- Because NOTIFY is broke, make sure our names are simple:
ALTER TABLE db      ADD CONSTRAINT db_name_sane      CHECK (name ~ E'^[a-zA-Z]\\w*$');
ALTER TABLE dbgroup ADD CONSTRAINT dbgroup_name_sane CHECK (name ~ E'^[a-zA-Z]\\w*$');
ALTER TABLE sync    ADD CONSTRAINT sync_name_sane    CHECK (name ~ E'^[a-zA-Z]\\w*$');


--
-- Track our children
--
CREATE SEQUENCE audit_pid_id_seq;
CREATE TABLE audit_pid (
  id             INTEGER     NOT NULL DEFAULT nextval('audit_pid_id_seq'),
  parentid       INTEGER         NULL, -- CTL or MCP id
  familyid       INTEGER         NULL, -- the MCP id
  type           TEXT        NOT NULL,
  sync           TEXT        NOT NULL,
  source         TEXT            NULL,
  target         TEXT            NULL,
  master_backend INT         NOT NULL DEFAULT pg_backend_pid(),
  source_backend INT             NULL,
  target_backend INT             NULL,
  ppid           INTEGER     NOT NULL,
  pid            INTEGER     NOT NULL,
  birthdate      TIMESTAMPTZ NOT NULL DEFAULT now(),
  killdate       TIMESTAMPTZ     NULL,
  birth          TEXT            NULL,
  death          TEXT            NULL
);

CREATE TABLE freezer.old_audit_pid AS SELECT * FROM audit_pid LIMIT 0;

--
-- Traffic control for children
--
CREATE TABLE q (
  sync      TEXT           NULL,
              CONSTRAINT q_sync_fk FOREIGN KEY (sync) REFERENCES sync(name) ON DELETE SET NULL,
  sourcedb  TEXT           NULL,
              CONSTRAINT q_sdb_fk FOREIGN KEY (sourcedb) REFERENCES db(name) ON DELETE SET NULL,
  targetdb  TEXT           NULL,
              CONSTRAINT q_tdb_fk FOREIGN KEY (targetdb) REFERENCES db(name) ON DELETE SET NULL,
  ppid      INTEGER     NOT NULL,
  pid       INTEGER         NULL,
  synctype  TEXT            NULL,
  updates   BIGINT          NULL,
  inserts   BIGINT          NULL,
  deletes   BIGINT          NULL,
  started   TIMESTAMPTZ     NULL,
  aborted   TIMESTAMPTZ     NULL,
  whydie    TEXT            NULL,
  ended     TIMESTAMPTZ     NULL,
  cdate     TIMESTAMPTZ NOT NULL DEFAULT now()
);
-- insert constraint - db and sync not null
-- Can only be one unstarted source->target per sync
CREATE UNIQUE INDEX "q_unique" ON q (sync,sourcedb,targetdb) WHERE started IS NULL;

CREATE INDEX q_ppid ON q (ppid,pid) WHERE ended IS NULL AND aborted IS NULL;

CREATE INDEX q_aborted ON q(sync) WHERE started IS NOT NULL AND aborted IS NOT NULL AND ended IS NULL;

CREATE INDEX q_cleanup ON q(cdate) WHERE ended IS NOT NULL;

CREATE INDEX q_stathelper ON q(cdate, sync) WHERE ended IS NOT NULL;


CREATE FUNCTION bucardo.bucardo_q()
RETURNS TRIGGER 
LANGUAGE plpgsql
AS $bc$
BEGIN
EXECUTE 'NOTIFY "bucardo_q_'||NEW.sync||'_'||NEW.targetdb||'"';
RETURN NEW;
END;
$bc$;

CREATE TRIGGER bucardo_q
  AFTER INSERT ON bucardo.q
  FOR EACH ROW EXECUTE PROCEDURE bucardo.bucardo_q();

CREATE TABLE freezer.master_q AS SELECT * FROM q LIMIT 0;
GRANT SELECT ON freezer.master_q TO PUBLIC;

CREATE FUNCTION bucardo_purge_q_table(interval)
RETURNS BIGINT
SECURITY DEFINER
LANGUAGE plpgsql
AS $bc$
DECLARE
  numrows BIGINT;
  qcount  BIGINT;
BEGIN
  RAISE DEBUG 'Purging q table of finished items older than %', $1;
  INSERT INTO freezer.master_q SELECT * FROM bucardo.q
    WHERE (ended IS NOT NULL OR aborted IS NOT NULL) AND cdate <= now() - $1;
  DELETE FROM bucardo.q WHERE (ended IS NOT NULL OR aborted IS NOT NULL) AND cdate <= now() - $1;
  GET DIAGNOSTICS numrows := row_count;
  SELECT count(*) FROM q INTO qcount;
  RAISE NOTICE 'Rows left in q table: %', qcount;
  RETURN numrows;
END;
$bc$;

CREATE FUNCTION bucardo_purge_q_table(text)
RETURNS BIGINT
LANGUAGE SQL
AS $bc$
SELECT bucardo_purge_q_table($1::interval);
$bc$;

CREATE FUNCTION bucardo_purge_q_table()
RETURNS BIGINT
LANGUAGE SQL
AS $bc$
SELECT bucardo_purge_q_table('2 hours'::interval);
$bc$;

CREATE FUNCTION bucardo_purge_q_table(integer)
RETURNS BIGINT
LANGUAGE plpgsql
IMMUTABLE
AS $bc$
BEGIN
RAISE EXCEPTION 'Please use an time interval such as ''2 hours''';
END;
$bc$;


CREATE FUNCTION table_exists(text,text)
RETURNS BOOLEAN
LANGUAGE plpgsql
AS $bc$
BEGIN
  PERFORM 1
    FROM pg_catalog.pg_class c, pg_namespace n
    WHERE c.relnamespace = n.oid
    AND n.nspname = $1
    AND c.relname = $2;
  IF FOUND THEN RETURN true; END IF;
  RETURN false;
END;
$bc$;

-- Called on update to master_q. Creates and populates child tables, empties master_q.
CREATE OR REPLACE FUNCTION populate_child_q_table()
RETURNS TRIGGER
LANGUAGE plpgsql
VOLATILE
SECURITY DEFINER
AS
$bc$
DECLARE
  myrec RECORD;
  myst  TEXT;
  needindex BOOL = false;
  tablename TEXT;
BEGIN

  -- Make sure we have all child tables
  FOR myrec IN SELECT DISTINCT TO_CHAR(cdate, 'YYYYMMDD') AS t FROM freezer.master_q LOOP
    tablename = 'child_q_' || myrec.t;
    RAISE DEBUG 'Found %', tablename;
    PERFORM 1 FROM pg_class c JOIN pg_namespace n ON (n.oid = c.relnamespace)
    WHERE n.nspname = 'freezer' AND c.relname = tablename;
    IF NOT FOUND THEN
      myst = 'CREATE TABLE freezer.' || tablename ||'() INHERITS (freezer.master_q)';
      EXECUTE myst;
      myst = 'GRANT SELECT ON freezer.' || tablename ||' TO public';
      EXECUTE myst;
      needindex = TRUE;
    END IF;

    IF needindex IS FALSE THEN
      PERFORM 1 FROM pg_class c JOIN pg_namespace n ON (n.oid = c.relnamespace)
      WHERE n.nspname = 'freezer' AND c.relname = tablename||'_daterange';
      IF NOT FOUND THEN
        needindex = TRUE;
      END IF;
    END IF;
    IF needindex IS TRUE THEN
      myst = 'CREATE INDEX '
             || tablename
             || '_daterange ON freezer.'
             || tablename
             || '(cdate)';
      EXECUTE myst;
    END IF;

    -- Move all the rows over!
    myst = 'INSERT INTO freezer.'
           || tablename
           || $$ SELECT * FROM ONLY freezer.master_q WHERE TO_CHAR(cdate,'YYYYMMDD') = $$
           || quote_literal(myrec.t);
    EXECUTE myst;
    myst = $$DELETE FROM ONLY freezer.master_q WHERE TO_CHAR(cdate, 'YYYYMMDD') = $$
           || quote_literal(myrec.t);
    EXECUTE myst;
  END LOOP;
  RETURN NULL;

END;
$bc$;

CREATE TRIGGER populate_child_q_table
  AFTER INSERT OR UPDATE ON freezer.master_q
  FOR EACH STATEMENT EXECUTE PROCEDURE bucardo.populate_child_q_table();

--
-- Return a created connection string from the db table
--
CREATE OR REPLACE FUNCTION bucardo.db_getconn(text)
RETURNS TEXT
LANGUAGE plperlu
SECURITY DEFINER
AS $bc$

## Given the name of a db, return a connection string, username, and password

use strict; use warnings; use DBI;
my ($name, $SQL, $rv, $row, %db);

$name = shift;

$name =~ s/'/''/go;
$SQL = "SELECT * FROM db WHERE name = '$name'";
$rv = spi_exec_query($SQL);
if (!$rv->{processed}) {
    elog(ERROR, qq{Error: Could not find a database with a name of $name\n});
}
$row = $rv->{rows}[0];

## If there is a dbfile and it exists, it overrides the rest
## Format = hostname:port:database:username:password
## http://www.postgresql.org/docs/current/static/libpq-pgpass.html

## We also check for one if no password is given
if (!defined $row->{dbpass}) {
    my $passfile = $row->{pgpass} || '';
    if (open my $pass, "<", $passfile) {
        ## We only do complete matches
        my $match = "$row->{dbhost}:$row->{dbport}:$row->{dbname}:$row->{dbuser}";
        while (<$pass>) {
            if (/^$match:(.+)/) {
                $row->{dbpass} = $1;
                elog(DEBUG, "Found password in pgpass file $passfile for $match");
                last;
            }
        }
    }
}

if (defined $row->{pgpass} and length $row->{pgpass}) {
    my $pfile = $row->{pgpass};
    if (! -e $pfile) {
        elog(WARNING, qq{Password file "$pfile" does not exist, falling back to default method.});
    }
    else {
        open my $pw, "<", $pfile or elog(ERROR, qq{Could not open password file "$pfile": $!\n});
        while(<$pw>) {
            next unless /^([^:]*):([^:]*):([^:]+):([^:]+):([^:]*)$/o;
            ($db{host},$db{port},$db{name},$db{user},$db{pw}) = ($1,$2,$3,$4,$5);
            last;
        }
        close $pw;
        exists $db{name} or elog(ERROR, qq{The password file "$pfile" did not contain any valid information\n});
    }
}

if (! exists $db{name}) {
    for (qw(host port name user pass conn)) {
        $db{$_} = exists $row->{"db$_"} ? $row->{"db$_"} : '';
    }
}

## Check that the port is numeric
if (defined $db{port} and length $db{port} and $db{port} !~ /^\d+$/) {
    elog(ERROR, qq{Database port must be numeric, but got "$db{port}"\n});
}

length $db{name} or elog(ERROR, qq{Database name is mandatory\n});
length $db{user} or elog(ERROR, qq{Database username is mandatory\n});

my $connstring = "dbi:Pg:dbname=$db{name}";
$db{host} ||= ''; $db{port} ||= ''; $db{pass} ||= '';
length $db{host} and $connstring .= ";host=$db{host}";
length $db{port} and $connstring .= ";port=$db{port}";
length $db{conn} and $connstring .= ";$db{conn}";

return "$connstring\n$db{user}\n$db{pass}";

$bc$;


--
-- Test a database connection, and log to the db_connlog table
--
CREATE FUNCTION bucardo.db_testconn(text)
RETURNS TEXT
LANGUAGE plperlu
SECURITY DEFINER
AS
$bc$

## Given the name of a db connection, construct the connection 
## string for it and then connect to it and log the attempt

use strict; use warnings; use DBI;
my ($name, $SQL, $rv, $row, $dbh, %db, $version, $found);

$name = shift;

$name =~ s/'/''/g;
$SQL = "SELECT bucardo.db_getconn('$name') AS bob";
$rv = spi_exec_query($SQL);
if (!$rv->{processed}) {
    elog(ERROR, qq{Error: Could not find a database with an name of $name\n});
}
$row = $rv->{rows}[0]{bob};
($db{dsn},$db{user},$db{pass}) = split /\n/ => $row;

elog(DEBUG, "Connecting as $db{dsn} user=$db{user} $$");

$dbh = DBI->connect($db{dsn}, $db{user}, $db{pass},
    {AutoCommit=>1, RaiseError=>1, PrintError=>0});

if (!$dbh) {
    elog(ERROR, qq{Database connection "$db{dsn}" failed: $DBI::errstr\n});
}

$version = $dbh->{pg_server_version};

# Install plpgsql if not there already
$SQL = q{SELECT 1 FROM pg_language WHERE lanname = 'plpgsql'};
my $sth = $dbh->prepare($SQL);
my $count = $sth->execute();
$sth->finish();
if ($count < 1) {
   $dbh->do("CREATE LANGUAGE plpgsql");
   $dbh->commit();
}

$dbh->disconnect();

my $safeconn = "$db{dsn} user=$db{user}"; ## No password for now
$safeconn =~ s/'/''/go;
$name =~ s/'/''/go;
$SQL = "INSERT INTO db_connlog (db,connstring,status,version) VALUES ('$name','$safeconn','good',$version)";
spi_exec_query($SQL);

return "Database connection successful";

$bc$;


--
-- Check the database connection if anything changes in the db table
--
CREATE FUNCTION db_change()
RETURNS TRIGGER
LANGUAGE plperlu
SECURITY DEFINER
AS
$bc$

return if $_TD->{new}{status} eq 'inactive';

## Test connection to the database specified
my $name = $_TD->{new}{name};
$name =~ s/'/''/g;
spi_exec_query("SELECT bucardo.db_testconn('$name')");
return;

$bc$;

CREATE TRIGGER db_change AFTER INSERT OR UPDATE ON db
  FOR EACH ROW EXECUTE PROCEDURE db_change();

--
-- Setup the goat table after any change
--
CREATE OR REPLACE FUNCTION validate_goat()
RETURNS TRIGGER
LANGUAGE plperlu
SECURITY DEFINER
AS
$bc$

## If a row in goat has changed, re-validate and set things up for that table
elog(DEBUG, "Running validate_goat");
use strict; use warnings; use DBI;

my ($SQL, $rv, $row, %db, $dbh, $sth, $count, $oid);

my $old = $_TD->{event} eq 'UPDATE' ? $_TD->{old} : 0;
my $new = $_TD->{new};

if (!defined $new->{db}) {
   die qq{Must provide a db\n};
}
if (!defined $new->{tablename}) {
   die qq{Must provide a tablename\n};
}
if (!defined $new->{schemaname}) {
   die qq{Must provide a schemaname\n};
}

my ($dbname,$schema,$table,$pkey) =
   ($new->{db}, $new->{schemaname}, $new->{tablename}, $new->{pkey});

## Do not allow pkeytype or qpkey to be set manually.
if (defined $new->{pkeytype} and (!$old or $new->{pkeytype} ne $old->{pkeytype})) {
    die qq{Cannot set pkeytype manually\n};
}
if (defined $new->{qpkey} and (!$old or $new->{qpkey} ne $old->{qpkey})) {
    die qq{Cannot set qpkey manually\n};
}

## If this is an update, we only continue if certain fields have changed
if ($old
    and $old->{db} eq $new->{db}
    and $old->{schemaname} eq $new->{schemaname}
    and $old->{tablename} eq $new->{tablename}
    and (defined $new->{pkey} and $new->{pkey} eq $old->{pkey})
    ) {
    #return;
}

(my $safedbname = $dbname) =~ s/'/''/go;
$SQL = "SELECT bucardo.db_getconn('$safedbname') AS apple";
$rv = spi_exec_query($SQL);
if (!$rv->{processed}) {
    elog(ERROR, qq{Error: Could not find a database with an name of $dbname\n});
}
$row = $rv->{rows}[0]{apple};
($db{dsn},$db{user},$db{pass}) = split /\n/ => $row;

elog(DEBUG, "Connecting in validate_goat as $db{dsn} user=$db{user} pid=$$ for table $schema.$table");

$dbh = DBI->connect($db{dsn}, $db{user}, $db{pass},
    {AutoCommit=>0, RaiseError=>1, PrintError=>0});

$dbh or elog(ERROR, qq{Database connection "$db{dsn}" as user $db{user} failed: $DBI::errstr\n});

## Get column information for this table (and verify it exists)
$SQL = q{
SELECT c.oid, attnum, attname, quote_ident(attname) AS qattname, typname, atttypid
FROM   pg_attribute a, pg_type t, pg_class c, pg_namespace n
WHERE  c.relnamespace = n.oid
AND    nspname = ? AND relname = ?
AND    a.attrelid = c.oid
AND    a.atttypid = t.oid
AND    attnum > 0
};
$sth = $dbh->prepare($SQL);
$count = $sth->execute($schema,$table);
if ($count < 1) {
   $sth->finish();
   $dbh->disconnect();
   die qq{Table not found at $db{dsn}: $schema.$table\n};
}
my $col = $sth->fetchall_hashref('attnum');
$oid = $col->{each %$col}{oid};

## Find all usable unique constraints for this table
$SQL = q{
SELECT   indisprimary, indkey
FROM     pg_index i
WHERE    indisunique AND indpred IS NULL AND indexprs IS NULL AND indrelid = ?
ORDER BY indexrelid DESC
};
## DESC because we choose the "newest" index in case of a tie below
$sth = $dbh->prepare($SQL);
$count = 0+$sth->execute($oid);
my $cons = $sth->fetchall_arrayref({});
$dbh->rollback();
$dbh->disconnect();

elog(DEBUG, "Valid unique constraints found: $count\n");
if ($count < 1) {
    ## We have no usable constraints. The entries must be blank.
    my $orignew = $new->{pkey};
    $new->{pkey} = $new->{qpkey} = $new->{pkeytype} = '';

    if (!$old) { ## This was an insert: just go
        elog(DEBUG, "No usable constraints, setting pkey et. al. to blank");
        return 'MODIFY';
    }

    ## If pkey has been set to NULL, this was a specific reset request, so return
    ## If pkey ended up blank (no change, or changed to blank), just return
    if (!defined $orignew or $orignew eq '') {
        return 'MODIFY';
    }

    ## The user has tried to change it something not blank, but this is not possible.
    die qq{Cannot set pkey for table $schema.$table: no unique constraint found\n};
}

## Pick the best possible one. Primary keys are always the best choice.
my ($primary)  = grep { $_->{indisprimary} } @$cons;
my $uniq;
if (defined $primary) {# and !$old and defined $new->{pkey}) {
    $uniq = $primary;
}
else {
    my (@foo) = grep { ! $_->{indisprimary} } @$cons;
    $count = @foo;
    ## Pick the one with the smallest number of columns.
    ## In case of a tie, choose the one with the smallest column footprint
    if ($count < 2) {
        $uniq = $foo[0];
    }
    else {
        my $lowest = 10_000;
        for (@foo) {
            my $cc = $_->{indkey} =~ y/ / /;
            if ($cc < $lowest) {
                $lowest = $cc;
                $uniq = $_;
            }
        }
    }
}

## This should not happen:
if (!defined $uniq) {
   die "Could not find a suitable unique index for table $schema.$table\n";
}

## If the user is not trying a manual override, set the best one and leave
if ((!defined $new->{pkey} or !length $new->{pkey}) or ($old and $new->{pkey} eq $old->{pkey})) {
    ($new->{pkey} = $uniq->{indkey}) =~ s/(\d+)(\s+)?/$col->{$1}{attname} . ($2 ? '|' : '')/ge;
    ($new->{qpkey} = $uniq->{indkey}) =~ s/(\d+)(\s+)?/$col->{$1}{qattname} . ($2 ? '|' : '')/ge;
    ($new->{pkeytype} = $uniq->{indkey}) =~ s/(\d+)(\s+)?/$col->{$1}{typname} . ($2 ? '|' : '')/ge;
    return 'MODIFY';
}

## They've attempted a manual override of pkey. Make sure it is valid.
for (@$cons) {
    (my $name = $_->{indkey}) =~ s/(\d+)(\s+)?/$col->{$1}{attname} . ($2 ? '|' : '')/ge;
    next unless $name eq $new->{pkey};
    ($new->{qpkey} = $uniq->{indkey}) =~ s/(\d+)(\s+)?/$col->{$1}{qattname} . ($2 ? '|' : '')/ge;
    ($new->{pkeytype} = $uniq->{indkey}) =~ s/(\d+)(\s+)?/$col->{$1}{typname} . ($2 ? '|' : '')/ge;
    return 'MODIFY';
}

die qq{Could not find a matching unique constraint that provides those columns\n};

$bc$; -- End of validate_goat()

CREATE TRIGGER validate_goat
  BEFORE INSERT OR UPDATE ON goat
  FOR EACH ROW EXECUTE PROCEDURE validate_goat();

--
-- Check that the goat tables are ready and compatible
--

CREATE OR REPLACE FUNCTION validate_sync(text,integer)
RETURNS TEXT
LANGUAGE plperlu
SECURITY DEFINER
AS
$bc$
    use strict; use warnings; use DBI;

    my $syncname = shift;
    my $force = shift || 0;
    my ($rv,$SQL,%cache);

    elog(NOTICE, "Starting validate_sync for $syncname");

    ## Connect to source and target(s) and verify that tables exist,
    ## and are identical. Setup the delta stuff as needed.

    ## Grab information about this sync from the database
    my $safename = $syncname;
    $safename =~ s/'/''/go;
    $SQL = "SELECT * FROM sync WHERE name = '$safename'";
    $rv = spi_exec_query($SQL);
    if (!$rv->{processed}) {
        elog(ERROR, "No such sync: $syncname");
    }

    my $info = $rv->{rows}[0];
    my $source = $info->{source};
    $source =~ s/'/''/go;
    ## Source is always a herd

    ## Prepare a list of all databases and tables involved in this sync
    ## We need to verify that the databases are reachable, that the tables exists,
    ## that the columns match up, and that the delta stuff is set up as needed.
    my %database;

    ## Does this herd exist?
    $SQL = qq{SELECT 1 FROM herd WHERE name = '$source'};
    $rv = spi_exec_query($SQL);
    if (!$rv->{processed}) {
       elog(ERROR, "No such herd: $source");
    }

    ## Process the source herd
    $SQL = qq{
        SELECT id, db, schemaname, tablename, pkey, pkeytype,
               standard_conflict, ping AS goatping,
               pg_catalog.quote_ident(db)         AS safedb,
               pg_catalog.quote_ident(schemaname) AS safeschema,
               pg_catalog.quote_ident(tablename)  AS safetable,
               pg_catalog.quote_ident(pkey)       AS safepkey
        FROM   goat g, herdmap h
        WHERE  g.id = h.goat
        AND    h.herd = '$source'
    };
    $rv = spi_exec_query($SQL);
    if (!$rv->{processed}) {
        elog(WARNING, "Herd has no members: $source");
        return qq{Herd "$source" for sync "$syncname" has no members: cannot validate};
    }

    ## All the same database, so we can extract it outside of the loop
    my $sourcedb = $rv->{rows}[0]{db};
    elog(DEBUG, "Got a sourcedb of $sourcedb for herd of $source");
    my %sourcetable;
    my %goat;
    for my $x (@{$rv->{rows}}) {
        $sourcetable{$x->{schemaname}}{$x->{tablename}} = $x;
    }

    ## Now to get all the target databases (will use schemas/tables from above)
    my $targetdb;
    if ($info->{targetdb}) {
        $targetdb->{$info->{targetdb}} = 'target';
    }
    elsif ($info->{targetgroup}) {
        my $group = $info->{targetgroup};
        $group =~ s/'/''/go;
        $SQL = qq{
            SELECT db, pg_catalog.quote_ident(db) AS safedb
            FROM   dbmap
            WHERE  dbgroup = '$group'
        };
        $rv = spi_exec_query($SQL);
        if (!@{$rv->{rows}}) {
            elog(ERROR, qq{Could not find a target database group of $info->{targetgroup}});
        }
        for (@{$rv->{rows}}) {
            $targetdb->{$_->{db}} = 'target';
        }
    }
    else {
        elog(ERROR, "Could not figure out the target for this sync!");
    }
    if (! keys %$targetdb) {
        elog(NOTICE, "No target databases found");
        return "No target databases foud for this sync";
    }

    ## We only want to check active databases
    my %dbstatus;
    $SQL = "SELECT name, status FROM db";
    $rv = spi_exec_query($SQL);
    for (@{$rv->{rows}}) {
        $dbstatus{$_->{name}} = $_->{status};
    }
    for (keys %$targetdb) {
        if ($dbstatus{$_} ne 'active') {
            elog(NOTICE, qq{Skipping inactive target database "$_". This may affect pruning of bucardo_delta});
            delete $targetdb->{$_};
        }
    }

    my $dbs = join "," => map { s/'/''/go; $_; } keys %$targetdb;
    if (!length $dbs) {
        elog(NOTICE, "No active databases found for this sync");
        return "No target databases found";
    }
    elog(DEBUG, "Target dbs: $dbs");

    ## If any of these are the source database, bail
    if (grep { $_ eq $sourcedb } keys %$targetdb) {
        elog(ERROR, "Source and target databases cannot be the same: $sourcedb");
    }

    ## Loop through and check each table in turn, setting up as needed.

    my %badtable;
    for my $schema (sort keys %sourcetable) {
        TABLE: for my $table (sort keys %{$sourcetable{$schema}}) {

            my $tinfo = $sourcetable{$schema}{$table};
            $tinfo->{synctype} = $info->{synctype};
            my $delta = $info->{synctype} =~ /pushdelta|swap/ ? 1 : 0;
            my $ping = $info->{ping} eq 't' ? 1 : 0;
            my $columns = bucardo_validate_table(1,$sourcedb,$tinfo,$delta,$ping,$syncname,\%cache,$source,$force);
            if ($columns == -1) {
                $badtable{$schema}{$table}=1;
                next TABLE;
            }

            $delta = $info->{synctype} =~ /swap/ ? 1 : 0;
            $ping = $delta;
            for my $tdb (sort keys %$targetdb) {
                my $tcolumns = bucardo_validate_table(0,$tdb,$tinfo,$delta,$ping,$syncname,\%cache,$source,$force);
                if ($columns == -1) {
                    $badtable{$schema}{$table}=1;
                    next TABLE;
                }
            } ## end each target database


        } ## end each source table

    }  ## end each source schema

    ## Remove any tables that were removed from the herd by bucardo_validate_table
    for my $schema (keys %badtable) {
        for my $table (keys %{$badtable{$schema}}) {
            delete $sourcetable{$schema}{$table};
        }
    }

    ## Update the bucardo_delta_targets table as needed
    my $check = "SELECT 1 FROM bucardo.bucardo_delta_targets WHERE tablename=? AND targetdb=?";
    my $add = "INSERT INTO bucardo.bucardo_delta_targets(tablename,targetdb) VALUES (?,?)";
    if ($info->{synctype} eq 'swap' or $info->{synctype} eq 'pushdelta') {
       ## Add all targets to the source
       my $dbh = $cache{dbh}{$sourcedb};
       my $sthc = $dbh->prepare($check);
       my $stha = $dbh->prepare($add);
       for my $tdb (sort keys %$targetdb) {
           for my $schema (sort keys %sourcetable) {
               for my $table (sort keys %{$sourcetable{$schema}}) {
                   my $toid = $cache{oid}{$sourcedb}{$schema}{$table};
                   my $ot = $cache{oid}{$tdb}{$schema}{$table};
                   elog(DEBUG, "Adding oid $toid to source, other is $ot for $schema.$table");
                   my $count = $sthc->execute($toid,$tdb);
                   $sthc->finish();
                   elog(DEBUG, "Checking on $tdb, count was $count");
                   if ($count != 1) {
                       $stha->execute($toid,$tdb);
                       $dbh->commit();
                   }
               }
           }
       }
    }
    if ($info->{synctype} eq 'swap') {
        ## Add source to the target(s)
        for my $tdb (sort keys %$targetdb) {
            my $dbh = $cache{dbh}{$tdb};
            my $sthc = $dbh->prepare($check);
            for my $schema (sort keys %sourcetable) {
                for my $table (sort keys %{$sourcetable{$schema}}) {
                    my $oid = $cache{oid}{$tdb}{$schema}{$table};
                    my $count = $sthc->execute($oid,$sourcedb);
                    $sthc->finish();
                    elog(DEBUG, "Checking on source $sourcedb, count was $count");
                    if ($count != 1) {
                        my $stha = $dbh->prepare($add);
                        $stha->execute($oid,$sourcedb);
                        $dbh->commit();
                    }
                }
            }
        }
    }

    sub run_sql {
        my ($sql,$dbh) = @_;
        if ($sql =~ /^(\s+)/m) {
            my $leading = length($1);
            $sql =~ s/^\s{$leading}//gsm;
            $sql =~ s/\t/    /gsm;
        }
        elog(DEBUG, "SQL: $sql");
        $dbh->do($sql);
    }

    sub bucardo_validate_table {

        my ($is_source, $db, $info, $delta, $ping, $syncname, $cache, $source, $force) = @_;
        my ($sth,$SQL);

        my ($schema, $safeschema, $table,  $safetable, $pkey, $safepkey, $safedb, $pkeytype) = @$info{qw(
           schemaname safeschema  tablename safetable   pkey   safepkey   safedb   pkeytype)};

        elog(DEBUG, "Validate_table for db $db $schema.$table, delta=$delta ping=$ping");

        $db =~ s/'/''/go;
        my $rv = spi_exec_query("SELECT bucardo.db_getconn('$db') AS conn");
        $rv->{processed} or elog(ERROR, qq{Error: Could not find a database named "$db"\n});
        my ($dsn,$user,$pass) = split /\n/ => $rv->{rows}[0]{conn};
        elog(DEBUG, "Connecting to $dsn as $user inside bucardo_validate_sync");
        my $dbh;
        if (exists $cache->{dbh}{$db}) {
            $dbh = $cache->{dbh}{$db};
            elog(DEBUG, 'Connected to cached version');
        }
        else {
            $cache->{dbh}{$db} = $dbh = DBI->connect($dsn, $user, $pass, {AutoCommit=>0, RaiseError=>1, PrintError=>0});
        }

        if ($info->{synctype} =~ /swap|delta/) {
            if (! $pkey) {
                if (!$force) {
                    elog(ERROR, qq{Table "$schema.$table" must specify a primary key when using a sync of '$info->{synctype}'});
                }
                warn qq{Table "$schema.$table" must specify a primary key when using a sync of '$info->{synctype}'};
                warn qq{REMOVING TABLE "$schema.$table" from herd "$source"\n};
                (my $safesource = $source) =~ s/'/''/g;
                $SQL = "DELETE FROM herdmap WHERE herd='$safesource' AND goat IN (SELECT id FROM goat ".
                "WHERE schemaname='$safeschema' AND tablename='$safetable')";
                $rv = spi_exec_query($SQL);
                return -1;
            }
            if (! $pkeytype) {
                elog(ERROR, qq{Table "$schema.$table" must specify a pkeytype when using a sync of '$info->{synctype}'});
            }
            if ($info->{synctype} =~ /swap/) {

                ## Grab a list of any custom conflict handlers
                $SQL = "SELECT 1 FROM customcode c, customcode_map m ".
                        "WHERE c.id=m.code AND m.goat=$info->{id} AND active IS TRUE ".
                        "AND whenrun = 'conflict'";
                elog(DEBUG, "Running $SQL");
                my $count = spi_exec_query($SQL)->{processed};
                if (! $info->{standard_conflict} and ! $count) {
                    elog(ERROR, qq{Table "$schema.$table" must specify a way to handle conflicts});
                }
            }
        }

        ## Get an inventory of supporting objects
        my %found;

        ## Make sure this schema/table combo exists!
        my $oid;
        $SQL = qq{
            SELECT c.oid
            FROM   pg_catalog.pg_class c, pg_catalog.pg_namespace n
            WHERE  c.relnamespace = n.oid
            AND    c.relkind = 'r'
            AND    n.nspname = ?
            AND    c.relname = ?
        };
        elog(DEBUG, "SQL: $SQL Args: $schema, $table");
        $sth = $dbh->prepare($SQL);
        my $count = $sth->execute($schema,$table);
        if ($count < 1) {
            $sth->finish();
            elog(ERROR, qq{No such table found for database $db: "$schema.$table"});
        }
        $oid = $sth->fetchall_arrayref()->[0][0];

        $cache->{oid}{$db}{$schema}{$table} = $oid;

        ## Make sure it does not have any verboten columns
        $SQL = qq{
            SELECT count(*)
            FROM   pg_catalog.pg_attribute
            WHERE  attrelid = $oid
            AND    NOT attisdropped
            AND    attname ~ '^BUCARDO';
        };
        $count = $dbh->selectall_arrayref($SQL)->[0][0];
        if ($count) {
            elog(ERROR, qq{Table "$schema.$table" contains a column starting with 'BUCARDO'});
        }

        $SQL = qq{
            SELECT count(*)
            FROM   pg_catalog.pg_namespace
            WHERE  nspname = ?
        };
        $sth = $dbh->prepare($SQL);
        $sth->execute('bucardo');
        $found{bc_schema} = $sth->fetchall_arrayref()->[0][0];

		## If we are not a delta sync, we can jump to the ping part
        my $func0 = length($syncname)    <= 42 ? "bucardo_triggerkick_$syncname" : "bkick_$syncname";
        my $trigger0 = "bucardo_triggerkick_$syncname";
		$delta or goto PINGCHECK;

        ## (if bc_schema exists) Do these triggers exist on this table?
        my $trigger1 = "bucardo_add_delta";
        my $trigger2 = "bucardo_add_delta_binary";

        if (!$found{bc_schema}) {
            $found{trigger0} = $found{trigger1} = $found{trigger2} = 0;
        }
        else {
            $SQL = qq{
                SELECT max(t0), max(t1), max(t2)
                FROM (
                SELECT
                  CASE WHEN t.tgname = ? THEN 1 ELSE 0 END AS t0,
                  CASE WHEN t.tgname = ? THEN 1 ELSE 0 END AS t1,
                  CASE WHEN t.tgname = ? THEN 1 ELSE 0 END AS t2
                FROM   pg_catalog.pg_trigger t, pg_catalog.pg_class c, pg_catalog.pg_namespace n
                WHERE  c.relnamespace = n.oid AND t.tgrelid = c.oid AND c.relkind = 'r'
                AND    n.nspname = ?
                AND    c.relname = ?
                AND    t.tgname  IN (?,?,?)
              ) foo
            };
            elog(DEBUG, "Preparing to check triggers: $SQL");
            $sth = $dbh->prepare($SQL);
            $sth->execute($trigger0,$trigger1,$trigger2,$schema,$table,$trigger0,$trigger1,$trigger2);
            my $x = 0;
            my $row = $sth->fetchall_arrayref()->[0];
            for my $val (@$row) {
                $found{"trigger$x"} = $val;
                $x++;
            }
        }

        ## (if bc_schema exists) Do these supporting functions exist?
        (my $noquotepkey = $safepkey) =~ s/^"(.+)"$/$1/;
        my $func1 = length($noquotepkey) <= 45 ? "bucardo_add_delta_$noquotepkey" : "bad_$noquotepkey";
        my $func2 = length($noquotepkey) <= 38 ? "bucardo_add_delta_binary_$noquotepkey" : "badb_$noquotepkey";
        my $func3 = "bucardo_purge_delta";
        my $func4 = "bucardo_compress_delta";
        if ($safepkey =~ /^"/) {
           $func1 = qq{"$func1"};
           $func2 = qq{"$func2"};
        }

        if (!$found{bc_schema}) {
            $found{func0} = $found{func1} = $found{func2} = $found{func3} = $found{func4} = 0;
        }
        else {
            $SQL = qq{
                SELECT max(f0), max(f1), max(f2), max(f3)
                FROM (
                SELECT
                  CASE WHEN p.proname = ? THEN 1 ELSE 0 END AS f0,
                  CASE WHEN p.proname = ? THEN 1 ELSE 0 END AS f1,
                  CASE WHEN p.proname = ? THEN 1 ELSE 0 END AS f2,
                  CASE WHEN p.proname = ? THEN 1 ELSE 0 END AS f3,
                  CASE WHEN p.proname = ? THEN 1 ELSE 0 END AS f4
                FROM   pg_catalog.pg_proc p, pg_catalog.pg_namespace n
                WHERE  p.pronamespace = n.oid
                AND    n.nspname = ?
                AND    p.proname IN (?,?,?,?,?)
            ) foo
            };
            elog(DEBUG, "Preparing to check functions: $SQL");
            $sth = $dbh->prepare($SQL);
            $sth->execute($func0,$func1,$func2,$func3,$func4,'bucardo',$func0,$func1,$func2,$func3,$func4);
            my $x = 0;
            my $row = $sth->fetchall_arrayref()->[0];
            for my $val (@$row) {
                $found{"func$x"} = $val;
                $x++;
            }
        }

        ## (if bc_schema exists) Does this table exist?

        if (!$found{bc_schema}) {
            $found{bucardo_delta} = $found{bucardo_track} = $found{bucardo_delta_targets} = 0;
        }
        else {
            $SQL = qq{
                SELECT max(t1), max(t2), max(t3)
                FROM (
                SELECT
                  CASE WHEN c.relname = 'bucardo_delta' THEN 1 ELSE 0 END AS t1,
                  CASE WHEN c.relname = 'bucardo_track' THEN 1 ELSE 0 END AS t2,
                  CASE WHEN c.relname = 'bucardo_delta_targets' THEN 1 ELSE 0 END AS t3
                FROM   pg_catalog.pg_class c, pg_catalog.pg_namespace n
                WHERE  c.relnamespace = n.oid
                AND    c.relkind = 'r'
                AND    n.nspname = 'bucardo'
                AND    c.relname IN ('bucardo_delta','bucardo_track','bucardo_delta_targets')
                ) foo
            };
            elog(DEBUG, "Preparing to check tables: $SQL");
            $sth = $dbh->prepare($SQL);
            $sth->execute();
            my $row = $sth->fetchall_arrayref()->[0];
            $found{'bucardo_delta'} = $row->[0];
            $found{'bucardo_track'} = $row->[1];
            $found{'bucardo_delta_targets'} = $row->[2];
        }

        ## (if bucardo_delta exists) Do these indexes exist?
        my $index1 = "bucardo_delta_${schema}_${table}_txn";
        my $l = length $index1;
        if ($l > 63) {
            if ($l <= 67) {
                $index1 = "bucardo_d_${schema}_${table}_txn";
            }
            elsif ($l <= 69) {
                $index1 = "bucardo_d_${schema}_${table}_t";
            }
            elsif ($l <= 76) {
                $index1 = "b_d_${schema}_${table}_t";
            }
            else {
                $index1 = "bucardo_delta_" . int(rand 9999999999);
            }
        }
        my $index2 = "bucardo_delta_${schema}_${table}_rowid";
        $l = length $index2;
        if ($l > 63) {
            if ($l <= 67) {
                $index2 = "bucardo_d_${schema}_${table}_rowid";
            }
            elsif ($l <= 71) {
                $index2 = "bucardo_d_${schema}_${table}_r";
            }
            elsif ($l <= 78) {
                $index2 = "b_d_${schema}_${table}_r";
            }
            else {
                $index2 = "bucardo_delta_" . int(rand 9999999999);
            }
        }
        my $index3 = "bucardo_track_target";
        my $index4 = "bucardo_delta_targetdb_unique";
        my $index5 = "bucardo_delta_txntime";
        my $safeindex1 = $index1;
        if ($safeindex1 =~ s/"/""/g or $safeindex1 =~ / /) {
            $safeindex1 = qq{"$safeindex1"};
        }

        if (!$found{bucardo_delta}) {
            $found{index1} = $found{index2} = $found{index3} = $found{index4} = $found{index5} = 0;
        }
        else {
            $SQL = qq{
                SELECT max(i1), max(i2), max(i3), max(i4), max(i5)
                FROM (
                SELECT
                  CASE WHEN c.relname = ? THEN 1 ELSE 0 END AS i1,
                  CASE WHEN c.relname = ? THEN 1 ELSE 0 END AS i2,
                  CASE WHEN c.relname = ? THEN 1 ELSE 0 END AS i3,
                  CASE WHEN c.relname = ? THEN 1 ELSE 0 END AS i4,
                  CASE WHEN c.relname = ? THEN 1 ELSE 0 END AS i5
                FROM   pg_catalog.pg_class c, pg_catalog.pg_namespace n
                WHERE  c.relnamespace = n.oid
                AND    c.relkind = 'i'
                AND    n.nspname = ?
                AND    c.relname IN (?,?,?,?,?)
                ) foo
            };
            elog(DEBUG, "Preparing to check indexes: $SQL");
            $sth = $dbh->prepare($SQL);
            $sth->execute($index1,$index2,$index3,$index4,$index5,'bucardo',$index1,$index2,$index3,$index4,$index5);
            my $x = 1;
            my $row = $sth->fetchall_arrayref()->[0];
            for my $val (@$row) {
                $found{"index$x"} = $val;
                $x++;
            }
        }

      PINGCHECK:
        ## Goat-level ping overrides the sync-level ping - source only
        if (defined $info->{goatping} and $is_source) {
            $ping = $info->{goatping} eq 't' ? 1 : 0;
            elog(DEBUG, "Got goatping as $info->{goatping}, force ping to $ping");
        }

        ## If this is not set as ping, remove the trigger - the only one safe to remove
        if (!$ping and $found{trigger0}) {
            run_sql(qq{DROP TRIGGER "$trigger0" ON $safeschema.$safetable}, $dbh);
        }

        ## Schema is needed for both ping and delta
        if (($ping or $delta) and !$found{bc_schema}) {
            run_sql('CREATE SCHEMA bucardo', $dbh);
        }

        ## Ping needs a special function and trigger
        if ($ping) {
            if (!$found{func0} or $force) {
                $SQL = qq{
                    CREATE OR REPLACE FUNCTION bucardo."$func0"()
                    RETURNS TRIGGER 
                    LANGUAGE plpgsql
                    AS \$notify\$
                    BEGIN
                        EXECUTE 'NOTIFY "bucardo_kick_sync_$syncname"';
                    RETURN NEW;
                    END;
                    \$notify\$;
                };
                run_sql($SQL,$dbh);
            }
            if (!$found{trigger0}) {
                $SQL = qq{
                    CREATE TRIGGER "$trigger0"
                    AFTER INSERT OR UPDATE OR DELETE ON $safeschema.$safetable
                    FOR EACH STATEMENT EXECUTE PROCEDURE bucardo."$func0"()
                };
                run_sql($SQL,$dbh);
            }
        }

        ## If this is not a delta, clean up what we can
        if (!$delta) {
            if ($found{bucardo_delta}) {
                elog(DEBUG, "Warning! NOT removing rows from bucardo.bucardo_delta where tablename=$oid ($table)");
            }
            if ($found{'bucardo_track'}) {
                elog(DEBUG, "Warning! NOT removing rows from bucardo.bucardo_track where tablename=$oid ($table)");
            }
            $found{index1} and elog(DEBUG, "Warning! NOT removing index bucardo.$safeindex1");
            $found{trigger1} and elog(DEBUG, "Warning! NOT removing trigger $trigger1 from $safeschema.$safetable");
            $found{trigger2} and elog(DEBUG, "Warning! NOT removing trigger $trigger2 from $safeschema.$safetable");

            ## Clean up the functions if nobody else is using them
            if ($found{func1} or $found{func2}) {
                $SQL = qq{
                    SELECT count(*)
                    FROM   pg_catalog.pg_trigger
                    WHERE  tgfoid = (
                        SELECT oid
                        FROM   pg_proc
                        WHERE  proname = ?
                        AND    pronamespace = (SELECT oid FROM pg_catalog.pg_namespace WHERE nspname = ?)
                    )
                };
                elog(DEBUG, "SQL: $SQL");
                $sth = $dbh->prepare($SQL);
                if ($found{func1}) {
                    $sth->execute($func1,'bucardo');
                    if (! $sth->fetchall_arrayref()->[0][0]) {
                        elog(DEBUG, "Warning! NOT removing function bucardo.$func1()");
                    }
                }
                if ($found{func2}) {
                    $sth = $dbh->prepare($SQL);
                    $sth->execute($func2,'bucardo');
                    if (! $sth->fetchall_arrayref()->[0][0]) {
                        elog(DEBUG, "Warning! NOT removing function bucardo.$func2()");
                    }
                }
            }
        } ## end if not delta
        else {

            my @pkeys = split (/\|/ => $pkey);
            my $numpkeys = @pkeys;

            if (! $found{bucardo_delta}) {
                $SQL = qq{
                    CREATE TABLE bucardo.bucardo_delta (
                        tablename  OID         NOT NULL,
                        rowid      TEXT        NOT NULL,
                        txntime    TIMESTAMPTZ NOT NULL DEFAULT now()
                    );
                };
                run_sql($SQL,$dbh);
                for (2..$numpkeys) {
                    $SQL = "ALTER TABLE bucardo.bucardo_delta ADD rowid$_ TEXT NULL";
                    run_sql($SQL,$dbh);
                }
            }

            if (! $found{'bucardo_track'}) {
                $SQL = qq{
                    CREATE TABLE bucardo.bucardo_track (
                        txntime    TIMESTAMPTZ NOT NULL,
                        tablename  OID         NOT NULL,
                        targetdb   TEXT        NOT NULL
                    );
                };
                run_sql($SQL,$dbh);
            }
            if (! $found{bucardo_delta_targets}) {
                $SQL = qq{
                    CREATE TABLE bucardo.bucardo_delta_targets (
                        tablename  OID         NOT NULL,
                        targetdb   TEXT        NOT NULL,
                        cdate      TIMESTAMPTZ NOT NULL DEFAULT now()
                    );
                };
                run_sql($SQL,$dbh);
            }
            if ($force) {
                if ($found{index1}) {
                    elog(DEBUG, qq{Dropping index "$safeindex1"});
                    $dbh->do(qq{DROP INDEX bucardo."$safeindex1"});
                    $found{index1} = 0;
                }
                if ($found{index2}) {
                    elog(DEBUG, qq{Dropping index "$index2"});
                    $dbh->do(qq{DROP INDEX bucardo."$index2"});
                    $found{index2} = 0;
                }
            }
            if (! $found{index1}) {
                $dbh->do(qq{CREATE INDEX "$safeindex1" ON bucardo.bucardo_delta(txntime) WHERE tablename = $oid});
            }
            for (2..$numpkeys) {
                my $colname = "rowid$_";
                $SQL = "SELECT count(*) FROM pg_catalog.pg_attribute a, pg_class c, pg_namespace n ".
                    "WHERE c.relnamespace = n.oid AND n.nspname = 'bucardo' AND a.attrelid = c.oid ".
                        "AND c.relname = 'bucardo_delta' AND a.attname = '$colname'";
                $count = $dbh->selectall_arrayref($SQL)->[0][0];
                if ($count != 1) {
                    $SQL = "ALTER TABLE bucardo.bucardo_delta ADD $colname TEXT NULL";
                    $dbh->do($SQL);
                }
            }
            if (! $found{index2}) {
                my @pkeytypes = split (/\|/ => $pkeytype);
                if (1 == @pkeytypes) {
                    my $safepkeytype = $pkeytype =~ /timestamp|timestamptz|date|bytea/ ? 'text' : $pkeytype;
                    elog(DEBUG, "Create index $index2 on bucardo_delta(rowid::$safepkeytype)");
                    $dbh->do(qq{CREATE INDEX "$index2" ON bucardo.bucardo_delta((rowid::$safepkeytype)) WHERE tablename = $oid});
                }
                else {
                    my $x = 0;
                    my $multicol = join ',' => map { s{timestamp|timestamptz|date|bytea}{text}; $x++; 
                        sprintf "(rowid%s::$_)", $x<2 ? '' : $x; } split (/\|/ => $pkeytype);
                    elog(DEBUG, "Creating index on $multicol");
                    $dbh->do(qq{CREATE INDEX "$index2" ON bucardo.bucardo_delta($multicol) WHERE tablename = $oid});
                }
            }
            if (! $found{index3}) {
                $dbh->do(qq{CREATE INDEX "$index3" ON bucardo.bucardo_track(tablename,txntime,targetdb)});
            }
            if (! $found{index4}) {
                $dbh->do(qq{CREATE UNIQUE INDEX "$index4" ON bucardo.bucardo_delta_targets(tablename,targetdb)});
            }
            if (! $found{index5}) {
                $dbh->do(qq{CREATE INDEX "$index5" ON bucardo.bucardo_delta(txntime)});
            }
            my $rowids = 'rowid';
            for (2..$numpkeys) {
                $rowids .= ",rowid$_";
            }
            my $new = join ',' => map { qq{NEW."$_"} } @pkeys;
            my $old = join ',' => map { qq{OLD."$_"} } @pkeys;
            my $clause = join ' OR ' => map { qq{OLD."$_" <> NEW."$_"} } @pkeys;
            if (! $found{func1} and $pkeytype ne 'bytea') {
                $SQL = qq{
                    CREATE OR REPLACE FUNCTION bucardo.$func1()
                    RETURNS TRIGGER
                    LANGUAGE plpgsql
                    SECURITY DEFINER
                    VOLATILE
                    AS
                    \$clone\$
                    BEGIN
                        IF (TG_OP = 'INSERT') THEN
                          INSERT INTO bucardo.bucardo_delta(tablename,$rowids) VALUES (TG_RELID, $new);
                        ELSIF (TG_OP = 'UPDATE') THEN
                          INSERT INTO bucardo.bucardo_delta(tablename,$rowids) VALUES (TG_RELID, $old);
                          IF ($clause) THEN
                            INSERT INTO bucardo.bucardo_delta(tablename,$rowids) VALUES (TG_RELID, $new);
                          END IF;
                        ELSE
                          INSERT INTO bucardo.bucardo_delta(tablename,$rowids) VALUES (TG_RELID, $old);
                        END IF;
                        RETURN NULL;
                    END;
                    \$clone\$;
                };
                run_sql($SQL,$dbh);
            }
            if (! $found{func2} and $pkeytype eq 'bytea') {
                $SQL = qq{
                    CREATE OR REPLACE FUNCTION bucardo.$func2()
                    RETURNS TRIGGER
                    LANGUAGE plpgsql
                    SECURITY DEFINER
                    VOLATILE
                    AS
                    \$clone\$
                    BEGIN
                        IF (TG_OP = 'INSERT') THEN
                          INSERT INTO bucardo.bucardo_delta(tablename,rowid) VALUES (TG_RELID,
                                   ENCODE(NEW.${safepkey},'base64'));
                        ELSIF (TG_OP = 'UPDATE') THEN
                          INSERT INTO bucardo.bucardo_delta(tablename,rowid) VALUES (TG_RELID,
                                   ENCODE(OLD.$safepkey,'base64'));
                          IF (ENCODE(OLD.$safepkey,'base64') <> ENCODE(NEW.$safepkey,'base64')) THEN
                            INSERT INTO bucardo.bucardo_delta(tablename,rowid) VALUES (TG_RELID,
                                   ENCODE(NEW.$safepkey,'base64'));
                          END IF;
                        ELSE
                          INSERT INTO bucardo.bucardo_delta(tablename,rowid) VALUES (TG_RELID,
                                   ENCODE(OLD.$safepkey,'base64'));
                        END IF;
                        RETURN NULL;
                    END;
                    \$clone\$;
                };
                run_sql($SQL,$dbh);
            }
            if (! $found{func3}) {
                $SQL = qq{
                    CREATE OR REPLACE FUNCTION bucardo.bucardo_purge_delta(interval)
                    RETURNS TEXT
                    LANGUAGE plpgsql
                    VOLATILE
                    SECURITY DEFINER
                    AS
                    \$clone\$
                    DECLARE
                        drows BIGINT;
                        trows BIGINT;
                    BEGIN
                        DELETE FROM bucardo.bucardo_delta
                        USING 
                        (SELECT b.tablename AS tn, b.txntime AS tt FROM
                        (SELECT tablename, count(*) FROM bucardo.bucardo_delta_targets GROUP BY 1) AS a,
                        (SELECT tablename, txntime, count(*) FROM bucardo.bucardo_track GROUP BY 1,2) AS b
                        WHERE a.tablename = b.tablename
                        AND a.count=b.count) AS foo
                        WHERE tablename = tn AND txntime = tt
                        AND txntime < now()-\$1;

                        GET DIAGNOSTICS drows := row_count;

                        DELETE FROM bucardo.bucardo_track
                        WHERE NOT EXISTS (SELECT 1 FROM bucardo.bucardo_delta d WHERE d.txntime = bucardo_track.txntime);

                        GET DIAGNOSTICS trows := row_count;

                        RETURN 'Rows deleted from bucardo_delta: '||drows||
                          ' Rows deleted from bucardo_track: '||trows;

                  END;
                    \$clone\$;
                };
                run_sql($SQL,$dbh);
            }
            if (! $found{func4}) {
                $SQL = qq{
                  CREATE OR REPLACE FUNCTION bucardo.bucardo_compress_delta(text, text)
                  RETURNS TEXT
                  LANGUAGE plpgsql
                  SECURITY DEFINER
                  AS \$clone\$
                  DECLARE
                    mymode TEXT;
                    myoid OID;
                    myst TEXT;
                    got2 bool;
                    drows BIGINT = 0;
                    trows BIGINT = 0;
                    rnames TEXT;
                    rname TEXT;
                    ids_where TEXT;
                    ids_sel TEXT;
                    ids_grp TEXT;
                    idnum TEXT;
                  BEGIN
                  
                  -- Are we running in serializable mode?
                  SELECT INTO mymode current_setting('transaction_isolation');
                  IF (mymode <> 'serializable') THEN
                    RAISE EXCEPTION 'This function must be run in serializable mode';
                  END IF;
                  
                  -- Grab the oid of this schema/table combo
                  SELECT INTO myoid
                    c.oid FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
                    WHERE nspname = \$1 AND relname = \$2;
                  
                  IF NOT FOUND THEN
                    RAISE EXCEPTION 'No such table: %.%', \$1, \$2;
                  END IF;
                  
                  ids_where = 'COALESCE(rowid,''NULL'') = COALESCE(id, ''NULL'')';
                  ids_sel = 'rowid AS id';
                  ids_grp = 'rowid';
                  FOR rname IN SELECT attname FROM pg_attribute WHERE attrelid =
                          (SELECT oid FROM pg_class WHERE relname = 'bucardo_delta'
                          AND relnamespace =
                          (SELECT oid FROM pg_namespace WHERE
                          nspname = 'bucardo') AND attname ~ '^rowid'
                      ) LOOP
                      rnames = COALESCE(rnames || ' ', '') || rname ;
                      SELECT INTO idnum SUBSTRING(rname FROM '[[:digit:]]+');
                      IF idnum IS NOT NULL THEN
                        ids_where = ids_where 
                          || ' AND (' 
                          || rname
                          || ' = id'
                          || idnum
                          || ' OR ('
                          || rname
                          || ' IS NULL AND id'
                          || idnum
                          || ' IS NULL))';
                        ids_sel = ids_sel
                          || ', '
                          || rname
                          || ' AS id'
                          || idnum;
                        ids_grp = ids_grp
                          || ', '
                          || rname;
                      END IF;
                  END LOOP;
                  
                  myst = 'DELETE FROM bucardo.bucardo_delta 
                  USING (SELECT MAX(txntime) AS maxt, '||ids_sel||'
                         FROM bucardo.bucardo_delta
                         WHERE tablename = '||myoid||'
                         GROUP BY ' || ids_grp || ') AS foo
                  WHERE tablename = '|| myoid || ' AND ' || ids_where ||' AND txntime <> maxt';
                  RAISE DEBUG 'Running %', myst;
                  EXECUTE myst;
                  
                  GET DIAGNOSTICS drows := row_count;
                  
                  myst = 'DELETE FROM bucardo.bucardo_track'
                    || ' WHERE NOT EXISTS (SELECT 1 FROM bucardo.bucardo_delta d WHERE d.txntime = bucardo_track.txntime)';
                  EXECUTE myst;
                  
                  GET DIAGNOSTICS trows := row_count;
                  
                  RETURN 'Compressed '||\$1||'.'||\$2||'. Rows deleted from bucardo_delta: '||drows||
                    ' Rows deleted from bucardo_track: '||trows;
                  
                  END;
                  \$clone\$;
                };
                run_sql($SQL,$dbh);

                $SQL = qq{
                  CREATE OR REPLACE FUNCTION bucardo.bucardo_compress_delta(text)
                  RETURNS TEXT
                  LANGUAGE SQL
                  SECURITY DEFINER
                  AS \$clone\$
                   SELECT bucardo.bucardo_compress_delta(n.nspname, c.relname) FROM pg_class c
                   JOIN pg_namespace n ON n.oid = c.relnamespace
                   WHERE relname = \$1 AND pg_table_is_visible(c.oid);
                  \$clone\$;
                };
                run_sql($SQL,$dbh);
                  
                $SQL = qq{
                  CREATE OR REPLACE FUNCTION bucardo.bucardo_compress_delta(oid)
                  RETURNS TEXT
                  LANGUAGE SQL
                  SECURITY DEFINER
                  AS \$clone\$
                   SELECT bucardo.bucardo_compress_delta(n.nspname, c.relname) FROM pg_class c
                   JOIN pg_namespace n ON n.oid = c.relnamespace
                   WHERE c.oid = \$1;
                  \$clone\$;
                };
                run_sql($SQL,$dbh);
                  
                $SQL = qq{
                  CREATE OR REPLACE FUNCTION bucardo.bucardo_compress_delta()
                  RETURNS SETOF TEXT
                  LANGUAGE SQL
                  SECURITY DEFINER
                  AS \$clone\$
                   SELECT bucardo.bucardo_compress_delta(n.nspname, c.relname) FROM pg_class c
                   JOIN pg_namespace n ON n.oid = c.relnamespace
                   WHERE c.oid IN (SELECT DISTINCT tablename FROM bucardo.bucardo_delta);
                  \$clone\$;
                };
                run_sql($SQL,$dbh);
            }
                  
            if (! $found{trigger1} and $pkeytype ne 'bytea') {
                $SQL = qq{
                    CREATE TRIGGER "$trigger1"
                    AFTER INSERT OR UPDATE OR DELETE ON $safeschema.$safetable
                    FOR EACH ROW EXECUTE PROCEDURE bucardo.$func1()
                };
                run_sql($SQL,$dbh);
            }
            if (! $found{trigger2} and $pkeytype eq 'bytea') {
                $SQL = qq{
                    CREATE TRIGGER "$trigger2"
                    AFTER INSERT OR UPDATE OR DELETE ON $safeschema.$safetable
                    FOR EACH ROW EXECUTE PROCEDURE bucardo.$func2()
                };
                run_sql($SQL,$dbh);
            }

        } ## end if delta

        $dbh->commit();
        return 1;

    } ## end of bucardo_validate_table

    for (values %{$cache{dbh}}) {
        $_->disconnect();
    }

    return 'MODIFY';
$bc$;

CREATE OR REPLACE FUNCTION bucardo.validate_sync(text)
RETURNS TEXT
LANGUAGE SQL
AS
$bc$
SELECT bucardo.validate_sync($1,0);
$bc$;

CREATE OR REPLACE FUNCTION bucardo.validate_all_syncs(integer)
RETURNS INTEGER
LANGUAGE plpgsql
AS
$bc$
DECLARE count INTEGER = 0; myrec RECORD;
BEGIN
FOR myrec IN SELECT name FROM sync ORDER BY name LOOP
  PERFORM validate_sync(myrec.name, $1);
  count = count + 1;
END LOOP;
RETURN count;
END;
$bc$;

CREATE OR REPLACE FUNCTION bucardo.validate_all_syncs()
RETURNS INTEGER
LANGUAGE SQL
AS
$bc$
SELECT validate_all_syncs(0);
$bc$;

CREATE FUNCTION validate_sync()
RETURNS TRIGGER
LANGUAGE plperlu
SECURITY DEFINER
AS $bc$

    use strict; use warnings;

    elog(DEBUG, "Starting validate_sync trigger");
    my $new = $_TD->{new};
    my $found=0;

    ## If insert, we always do the full validation:

    if ($_TD->{event} eq 'INSERT') {
        elog(DEBUG, "Found insert, will call validate_sync");
        $found = 1;
    }
    else {
        my $old = $_TD->{old};
        for my $x (qw(name source targetdb targetgroup synctype ping)) {
            elog(DEBUG, "Checking on $x");
            if (! defined $old->{$x}) {
                next if ! defined $new->{$x};
            }
            elsif (defined $new->{$x} and $new->{$x} eq $old->{$x}) {
                next;
            }
            $found=1;
            last;
        }
    }
    if ($found) {
        spi_exec_query("SELECT validate_sync('$new->{name}')");
    }
    return;
$bc$;

CREATE TRIGGER validate_sync
  AFTER INSERT OR UPDATE ON sync
  FOR EACH ROW EXECUTE PROCEDURE validate_sync();

CREATE FUNCTION bucardo_delete_sync()
RETURNS TRIGGER
LANGUAGE plperlu
SECURITY DEFINER
AS $bc$

    use strict; use warnings;

    elog(DEBUG, "Starting delete_sync trigger");

    my $old = $_TD->{old};

    ## If this sync was fullcopy, we don't need to worry about it
    return if $old->{synctype} eq 'fullcopy';

    my ($SQL, $rv, $sth, $count);

    ## Gather up a list of tables used in this sync, as well as the source database handle

    (my $herd = $old->{source}) =~ s/'/''/go;
    $SQL = qq{
        SELECT db, pg_catalog.quote_ident(schemaname) AS safeschema,
                   pg_catalog.quote_ident(tablename)  AS safetable
        FROM   goat g, herdmap h
        WHERE  g.id = h.goat
        AND    h.herd = '$herd'
    };
    $rv = spi_exec_query($SQL);
    $rv->{processed} or die qq{No such herd: $herd};
    my $sourcedb = $rv->{rows}[0]{db};
    elog(DEBUG, "Got a sourcedb of $sourcedb for herd of $herd");
    my %relation;
    for (@{$rv->{rows}}) {
        $relation{$_->{safeschema}}{$_->{safetable}} = $_;
    }
    $rv = spi_exec_query("SELECT bucardo.db_getconn('$sourcedb') AS conn");
    $rv->{processed} or die qq{Could not find a database named "$sourcedb"};
    my ($dsn,$user,$pass) = split /\n/ => $rv->{rows}[0]{conn};
    elog(DEBUG, "Connecting to $dsn as $user inside bucardo_delete_sync");
    my $sdbh = DBI->connect($dsn, $user, $pass, {AutoCommit=>0, RaiseError=>1, PrintError=>0});

    ## Get the list of target databases

    my %target;
    if ($old->{targetdb}) {
       $target{$old->{targetdb}} = 1;
    }
    else {
        my $group = $old->{targetgroup};
        $group =~ s/'/''/g;
        $SQL = "SELECT db FROM dbmap WHERE dbgroup = '$group'";
        $rv = spi_exec_query($SQL);
        $rv->{processed} or die qq{Could not find the dbgroup $group};
        for (@{$rv->{rows}}) {
            $target{$_->{db}} = 1;
        }
    }

    ## Try and delete each combo
    $SQL = "DELETE FROM bucardo.bucardo_delta_targets WHERE targetdb = ? AND tablename = ".
        "(SELECT c.oid FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n WHERE c.relname = ? ".
        "AND c.relnamespace = n.oid AND n.nspname = ?)";
    $sth = $sdbh->prepare($SQL);
    for my $schema (sort keys %relation) {
        for my $table (sort keys %{$relation{$schema}}) {
            for my $target (keys %target) {
                $count = $sth->execute($target,$table,$schema);
                elog(DEBUG,"Tried to remove $schema:$table for $target, got $count");
            }
        }
    }
    $sdbh->commit();

    return if $old->{synctype} eq 'pushdelta';

    ## Finally, remove from the target database
    (my $targetdb = $old->{targetdb}) =~ s/'/''/go;
    $rv = spi_exec_query("SELECT bucardo.db_getconn('$targetdb') AS conn");
    $rv->{processed} or die qq{Could not find a database named "$targetdb"};
    ($dsn,$user,$pass) = split /\n/ => $rv->{rows}[0]{conn};
    elog(DEBUG, "Connecting to $dsn as $user inside bucardo_delete_sync");
    my $tdbh = DBI->connect($dsn, $user, $pass, {AutoCommit=>0, RaiseError=>1, PrintError=>0});

    $sth = $tdbh->prepare($SQL);
    for my $schema (sort keys %relation) {
        for my $table (sort keys %{$relation{$schema}}) {
            $count = $sth->execute($sourcedb,$table,$schema);
            elog(DEBUG,"Tried to remove $schema:$table for $sourcedb, got $count");
        }
    }
    $tdbh->commit();

    return;

$bc$;

CREATE TRIGGER bucardo_delete_sync
  AFTER DELETE ON sync
  FOR EACH ROW EXECUTE PROCEDURE bucardo_delete_sync();

CREATE SEQUENCE customcode_id_seq;
CREATE TABLE customcode (
  id       INTEGER     NOT NULL DEFAULT nextval('customcode_id_seq'),
             CONSTRAINT customcode_id_pk PRIMARY KEY (id),
  name     TEXT        NOT NULL UNIQUE,
  about    TEXT            NULL,
  whenrun  TEXT        NOT NULL,
  getdbh   BOOLEAN     NOT NULL DEFAULT 'true',
  getrows  BOOLEAN     NOT NULL DEFAULT 'false',
  src_code TEXT        NOT NULL,
  cdate    TIMESTAMPTZ NOT NULL DEFAULT now()
);
ALTER TABLE customcode ADD CONSTRAINT customcode_whenrun
CHECK (whenrun IN (
'before_txn',
'before_check_rows',
'before_trigger_drop',
'after_trigger_drop',
'after_table_sync',
'exception',
'conflict',
'before_trigger_enable',
'after_trigger_enable',
'after_txn',
'before_sync',
'after_sync'
));

CREATE TABLE customcode_map (
  code     INTEGER     NOT NULL,
             CONSTRAINT customcode_map_code_fk FOREIGN KEY (code) REFERENCES customcode(id) ON DELETE CASCADE,
  sync     TEXT            NULL,
             CONSTRAINT customcode_map_sync_fk FOREIGN KEY (sync) REFERENCES sync(name) ON DELETE SET NULL,
  goat     INTEGER         NULL,
             CONSTRAINT customcode_map_goat_fk FOREIGN KEY (goat) REFERENCES goat(id) ON DELETE SET NULL,
  active   BOOLEAN     NOT NULL DEFAULT 'true',
  priority SMALLINT    NOT NULL DEFAULT 0,
  cdate    TIMESTAMPTZ NOT NULL DEFAULT now()
);
ALTER TABLE customcode_map ADD CONSTRAINT customcode_map_syncgoat
CHECK (sync IS NULL OR goat IS NULL);

CREATE UNIQUE INDEX customcode_map_unique_sync ON customcode_map(code,sync) WHERE sync IS NOT NULL;
CREATE UNIQUE INDEX customcode_map_unique_goat ON customcode_map(code,goat) WHERE goat IS NOT NULL;


CREATE OR REPLACE FUNCTION find_unused_goats()
RETURNS SETOF text
LANGUAGE plpgsql
AS $bc$
DECLARE
  myrec RECORD;
BEGIN
  FOR myrec IN 
    SELECT quote_ident(db) || '.' || quote_ident(schemaname) || '.' || quote_ident(tablename) AS t
      FROM goat g
      WHERE NOT EXISTS (SELECT 1 FROM herdmap h WHERE h.goat = g.id)
      ORDER BY schemaname, tablename
    LOOP
      RETURN NEXT 'Not used in any herds: ' || myrec.t;
  END LOOP;

  FOR myrec IN 
    SELECT quote_ident(db) || '.' || quote_ident(schemaname) || '.' || quote_ident(tablename) AS t
      FROM goat g
      JOIN herdmap h ON h.goat = g.id
      WHERE NOT EXISTS (SELECT 1 FROM sync WHERE source = h.herd)
      ORDER BY schemaname, tablename
    LOOP
      RETURN NEXT 'Not used in source herd: ' || myrec.t;
  END LOOP;

  FOR myrec IN 
    SELECT quote_ident(db) || '.' || quote_ident(schemaname) || '.' || quote_ident(tablename) AS t
      FROM goat g
      JOIN herdmap h ON h.goat = g.id
      WHERE NOT EXISTS (SELECT 1 FROM sync WHERE source = h.herd AND status = 'active')
      ORDER BY schemaname, tablename
    LOOP
      RETURN NEXT 'Not used in source herd of active sync: ' || myrec.t;
  END LOOP;

  RETURN;
END;
$bc$;

-- Monitor how long data takes to move over, from commit to commit
CREATE TABLE bucardo_rate (
  sync         TEXT        NOT NULL,
  goat         INTEGER     NOT NULL,
  target       TEXT            NULL,
  mastercommit TIMESTAMPTZ NOT NULL,
  slavecommit  TIMESTAMPTZ NOT NULL,
  total        INTEGER     NOT NULL
);
CREATE INDEX bucardo_rate_sync ON bucardo_rate(sync);

-- Keep track of any upgrades as we go along
CREATE TABLE upgrade_log (
  action   TEXT        NOT NULL,
  summary  TEXT        NOT NULL,
  version  TEXT        NOT NULL,
  cdate    TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Allow users to insert messages in the Bucardo logs

CREATE FUNCTION bucardo_log_message_notify()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $bc$
BEGIN
  EXECUTE 'NOTIFY "bucardo_log_message"';
  RETURN NULL;
END;
$bc$;

CREATE TABLE bucardo_log_message (
  msg TEXT NOT NULL,
  cdate TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TRIGGER bucardo_log_message_trigger
  AFTER INSERT ON bucardo_log_message
  FOR EACH STATEMENT EXECUTE PROCEDURE bucardo_log_message_notify();


--
-- END OF THE SCHEMA
--