Pg Table Sync

gistfile1.sql

-- DROP FUNCTION table_sync(regclass, regclass);
CREATE OR REPLACE FUNCTION table_sync(in_source regclass, in_destination regclass)
  RETURNS void AS
$BODY$DECLARE
        _pk name[];
        _data name[];
        _all name[];

        _pk_ident text;
        _data_ident text;
        _data_set_sql text;
        _pk_match_sql text;
        _data_ident_dst text;
        _data_ident_src text;
        _all_ident text;

        _sql text;

        _rows bigint;
BEGIN

_pk := ARRAY(
        SELECT attname
        FROM pg_catalog.pg_constraint c, pg_catalog.pg_attribute ac
        WHERE
                c.contype = 'p'
                AND c.conrelid = in_destination::oid
                AND ac.attnum = ANY(c.conkey)
                AND c.conrelid = ac.attrelid
                AND ac.attnum > 0
                AND NOT ac.attisdropped
);

_data := ARRAY(
        SELECT attname
        FROM pg_catalog.pg_constraint c, pg_catalog.pg_attribute ac
        WHERE
                c.contype = 'p'
                AND c.conrelid = in_destination::oid
                AND NOT ac.attnum = ANY(c.conkey)
                AND c.conrelid = ac.attrelid
                AND ac.attnum > 0
                AND NOT ac.attisdropped
);

_all := ARRAY(
        SELECT attname
        FROM pg_catalog.pg_attribute a
        WHERE
                a.attrelid = in_destination::oid
                AND a.attnum > 0
                AND NOT a.attisdropped
);

_pk_ident := array_to_string(ARRAY(
        SELECT quote_ident(attr)
        FROM unnest(_pk) AS attr
), ', ');

_data_ident := array_to_string(ARRAY(
        SELECT quote_ident(attr)
        FROM unnest(_data) AS attr
), ', ');

_all_ident := array_to_string(ARRAY(
        SELECT quote_ident(attr)
        FROM unnest(_all) AS attr
), ', ');

_data_ident_src := array_to_string(ARRAY(
        SELECT 'src.' || quote_ident(attr)
        FROM unnest(_data) AS attr
), ', ');

_data_ident_dst := array_to_string(ARRAY(
        SELECT 'dst.' || quote_ident(attr)
        FROM unnest(_data) AS attr
), ', ');

_pk_match_sql := array_to_string(ARRAY(
        SELECT 'src.' || quote_ident(attr) || ' = dst.' || quote_ident(attr)
        FROM unnest(_pk) AS attr
), ' AND ');

_data_set_sql := array_to_string(ARRAY(
        SELECT quote_ident(attr) || '=src.' || quote_ident(attr)
        FROM unnest(_data) AS attr
), ', ');

_sql := $$DELETE FROM $$ || quote_ident(in_destination::text) || $$
WHERE ($$ || _pk_ident || $$) NOT IN (
        SELECT $$ || _pk_ident || $$
        FROM $$ || quote_ident(in_source::text) || $$
)$$;

EXECUTE _sql;
GET DIAGNOSTICS _rows = ROW_COUNT;
RAISE INFO 'rows deleted: %', _rows;


_sql := $$UPDATE $$ || quote_ident(in_destination::text) || $$ AS dst
SET $$ || _data_set_sql || $$
FROM $$ || quote_ident(in_source::text) || $$ AS src
WHERE
        $$ || _pk_match_sql || $$
        AND ($$ || _data_ident_src || $$) IS DISTINCT FROM ($$ || _data_ident_dst || $$)$$;

EXECUTE _sql;
GET DIAGNOSTICS _rows = ROW_COUNT;
RAISE INFO 'rows updated: %', _rows;


_sql := $$INSERT INTO $$ || quote_ident(in_destination::text) || ' (' || _all_ident || $$)
SELECT $$ || _all_ident || $$
FROM $$ || quote_ident(in_source::text) || $$
WHERE ($$ || _pk_ident || $$) NOT IN (
        SELECT $$ || _pk_ident || $$
        FROM $$ || quote_ident(in_destination::text) || $$
)$$;

EXECUTE _sql;
GET DIAGNOSTICS _rows = ROW_COUNT;
RAISE INFO 'rows inserted: %', _rows;

END;
$BODY$
  LANGUAGE plpgsql VOLATILE STRICT;



-- DROP FUNCTION table_checksum(regclass);
CREATE OR REPLACE FUNCTION table_checksum(in_table regclass, in_reference regclass)
  RETURNS character AS
$BODY$DECLARE
        _pk name[];
        _all name[];

        _pk_ident text;
        _all_ident text;

        _result character(32);
BEGIN

_pk := ARRAY(
        SELECT attname
        FROM pg_catalog.pg_constraint c, pg_catalog.pg_attribute ac
        WHERE
                c.contype = 'p'
                AND c.conrelid = in_reference::oid
                AND ac.attnum = ANY(c.conkey)
                AND c.conrelid = ac.attrelid
                AND ac.attnum > 0
                AND NOT ac.attisdropped
);

_all := ARRAY(
        SELECT attname
        FROM pg_catalog.pg_attribute a
        WHERE
                a.attrelid = in_reference::oid
                AND a.attnum > 0
                AND NOT a.attisdropped
);

_pk_ident := array_to_string(ARRAY(
        SELECT quote_ident(attr)
        FROM unnest(_pk) AS attr
), ', ');

_all_ident := array_to_string(ARRAY(
        SELECT quote_ident(attr)
        FROM unnest(_all) AS attr
), ', ');



EXECUTE $$
        SELECT md5(array_to_string(ARRAY(
                SELECT md5(($$ || _all_ident || $$)::text)
                FROM $$ || quote_ident(in_table::text) || $$
                ORDER BY $$ || _pk_ident || $$
        ), ''))
$$ INTO _result;

RETURN _result;

END;
$BODY$
  LANGUAGE plpgsql VOLATILE STRICT;