-- PostRR - src/postrr.sql
-- Copyright (C) 2012 Sebastian 'tokkee' Harl <sh@tokkee.org>
-- All rights reserved.
--
-- Redistribution and use in source and binary forms, with or without
-- modification, are permitted provided that the following conditions
-- are met:
-- 1. Redistributions of source code must retain the above copyright
--    notice, this list of conditions and the following disclaimer.
-- 2. Redistributions in binary form must reproduce the above copyright
--    notice, this list of conditions and the following disclaimer in the
--    documentation and/or other materials provided with the distribution.
--
-- THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
-- ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
-- TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
-- PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDERS OR
-- CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
-- EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
-- PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
-- OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
-- WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
-- OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
-- ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

--
-- PostRR - PostgreSQL Round-Robin Extension
--

-- complain if script is sourced in psql, rather than via CREATE EXTENSION
\echo Use "CREATE EXTENSION postrr" to load this file. \quit

CREATE SCHEMA postrr;

CREATE SEQUENCE postrr.tsid INCREMENT BY 1
	MINVALUE 1 MAXVALUE 2147483647 START WITH 1 NO CYCLE;

CREATE TABLE postrr.rrtimeslices (
	tsid integer NOT NULL PRIMARY KEY
		DEFAULT nextval('postrr.tsid'::regclass)
		CHECK (0 < tsid),
	tslen integer NOT NULL,
	tsnum integer NOT NULL
);

SELECT pg_catalog.pg_extension_config_dump('postrr.rrtimeslices', '');

CREATE TABLE postrr.rrarchives (
	rraname text NOT NULL,
	tbl name NOT NULL,
	tscol name NOT NULL,
	vcol name NOT NULL,
	UNIQUE (rraname, tbl, tscol, vcol)
);

SELECT pg_catalog.pg_extension_config_dump('postrr.rrarchives', '');

CREATE OR REPLACE FUNCTION PostRR_Version()
	RETURNS cstring
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'postrr_version'
	LANGUAGE C IMMUTABLE;

CREATE TYPE RRTimeslice;

CREATE OR REPLACE FUNCTION RRTimeslice_validate(integer)
	RETURNS cstring
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_validate'
	LANGUAGE C IMMUTABLE STRICT;

-- this will abort the transaction in case the expected internal length does
-- not match the actual length
SELECT RRTimeslice_validate(16);

CREATE OR REPLACE FUNCTION RRTimeslice_in(cstring, oid, integer)
	RETURNS RRTimeslice
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_in'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION RRTimeslice_out(RRTimeslice)
	RETURNS cstring
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_out'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION RRTimeslice_typmodin(cstring[])
	RETURNS integer
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_typmodin'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION RRTimeslice_typmodout(integer)
	RETURNS cstring
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_typmodout'
	LANGUAGE C IMMUTABLE STRICT;

CREATE TYPE RRTimeslice (
	INTERNALLENGTH = 16,
	INPUT          = RRTimeslice_in,
	OUTPUT         = RRTimeslice_out,
	TYPMOD_IN      = RRTimeslice_typmodin,
	TYPMOD_OUT     = RRTimeslice_typmodout,
	ALIGNMENT      = double,
	STORAGE        = plain
);

CREATE OR REPLACE FUNCTION RRTimeslice(rrtimeslice, integer, boolean)
	RETURNS rrtimeslice
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_to_rrtimeslice'
	LANGUAGE C IMMUTABLE STRICT;

CREATE CAST (rrtimeslice AS rrtimeslice)
	WITH FUNCTION RRTimeslice(rrtimeslice, integer, boolean)
	AS IMPLICIT;

CREATE OR REPLACE FUNCTION RRTimeslice(timestamptz, integer, boolean)
	RETURNS rrtimeslice
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'timestamptz_to_rrtimeslice'
	LANGUAGE C IMMUTABLE STRICT;

CREATE CAST (timestamptz AS rrtimeslice)
	WITH FUNCTION RRTimeslice(timestamptz, integer, boolean)
	AS IMPLICIT;

CREATE OR REPLACE FUNCTION Tstamptz(rrtimeslice)
	RETURNS timestamptz
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_to_timestamptz'
	LANGUAGE C IMMUTABLE STRICT;

CREATE CAST (rrtimeslice AS timestamptz)
	WITH FUNCTION Tstamptz(rrtimeslice);
	-- EXPLICIT

CREATE OR REPLACE FUNCTION rrtimeslice_cmp(rrtimeslice, rrtimeslice)
	RETURNS integer
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_cmp'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION rrtimeslice_seq_eq(rrtimeslice, rrtimeslice)
	RETURNS boolean
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_seq_eq'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION rrtimeslice_seq_ne(rrtimeslice, rrtimeslice)
	RETURNS boolean
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_seq_ne'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION rrtimeslice_seq_lt(rrtimeslice, rrtimeslice)
	RETURNS boolean
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_seq_lt'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION rrtimeslice_seq_le(rrtimeslice, rrtimeslice)
	RETURNS boolean
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_seq_le'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION rrtimeslice_seq_gt(rrtimeslice, rrtimeslice)
	RETURNS boolean
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_seq_gt'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION rrtimeslice_seq_ge(rrtimeslice, rrtimeslice)
	RETURNS boolean
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_seq_ge'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION rrtimeslice_seq_cmp(rrtimeslice, rrtimeslice)
	RETURNS integer
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_seq_cmp'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION rrtimeslice_seq_hash(rrtimeslice)
	RETURNS integer
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'rrtimeslice_seq_hash'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OPERATOR = (
	LEFTARG    = RRTimeslice,
	RIGHTARG   = RRTimeslice,
	PROCEDURE  = rrtimeslice_seq_eq,
	COMMUTATOR = =,
	NEGATOR    = <>,
	RESTRICT   = eqsel
);

CREATE OPERATOR <> (
	LEFTARG    = RRTimeslice,
	RIGHTARG   = RRTimeslice,
	PROCEDURE  = rrtimeslice_seq_ne,
	COMMUTATOR = <>,
	NEGATOR    = =,
	RESTRICT   = neqsel
);

CREATE OPERATOR < (
	LEFTARG    = RRTimeslice,
	RIGHTARG   = RRTimeslice,
	PROCEDURE  = rrtimeslice_seq_lt,
	COMMUTATOR = >,
	NEGATOR    = <=,
	RESTRICT   = scalarltsel
);

CREATE OPERATOR <= (
	LEFTARG    = RRTimeslice,
	RIGHTARG   = RRTimeslice,
	PROCEDURE  = rrtimeslice_seq_le,
	COMMUTATOR = >=,
	NEGATOR    = <,
	RESTRICT   = scalarltsel
);

CREATE OPERATOR > (
	LEFTARG    = RRTimeslice,
	RIGHTARG   = RRTimeslice,
	PROCEDURE  = rrtimeslice_seq_gt,
	COMMUTATOR = <,
	NEGATOR    = >=,
	RESTRICT   = scalargtsel
);

CREATE OPERATOR >= (
	LEFTARG    = RRTimeslice,
	RIGHTARG   = RRTimeslice,
	PROCEDURE  = rrtimeslice_seq_ge,
	COMMUTATOR = <=,
	NEGATOR    = >,
	RESTRICT   = scalargtsel
);

CREATE OPERATOR CLASS rrtimeslice_ops
	DEFAULT FOR TYPE RRTimeslice USING btree AS
		OPERATOR 1 < ,
		OPERATOR 2 <= ,
		OPERATOR 3 = ,
		OPERATOR 4 >= ,
		OPERATOR 5 > ,
		FUNCTION 1 rrtimeslice_seq_cmp(rrtimeslice, rrtimeslice);

CREATE OPERATOR CLASS rrtimeslice_hash_ops
	FOR TYPE RRTimeslice USING hash AS
		OPERATOR 1 = ,
		FUNCTION 1 rrtimeslice_seq_hash(rrtimeslice);

CREATE TYPE CData;

CREATE OR REPLACE FUNCTION CData_validate(integer)
	RETURNS cstring
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'cdata_validate'
	LANGUAGE C IMMUTABLE STRICT;

-- this will abort the transaction in case the expected internal length does
-- not match the actual length
SELECT CData_validate(24);

CREATE OR REPLACE FUNCTION CData_in(cstring, oid, integer)
	RETURNS CData
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'cdata_in'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION CData_out(CData)
	RETURNS cstring
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'cdata_out'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION CData_typmodin(cstring[])
	RETURNS integer
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'cdata_typmodin'
	LANGUAGE C IMMUTABLE STRICT;

CREATE OR REPLACE FUNCTION CData_typmodout(integer)
	RETURNS cstring
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'cdata_typmodout'
	LANGUAGE C IMMUTABLE STRICT;

CREATE TYPE CData (
	INTERNALLENGTH = 24,
	INPUT          = CData_in,
	OUTPUT         = CData_out,
	TYPMOD_IN      = CData_typmodin,
	TYPMOD_OUT     = CData_typmodout,
	ALIGNMENT      = double,
	STORAGE        = plain
);

CREATE OR REPLACE FUNCTION CData(cdata, integer, boolean)
	RETURNS cdata
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'cdata_to_cdata'
	LANGUAGE C IMMUTABLE STRICT;

CREATE CAST (cdata AS cdata)
	WITH FUNCTION CData(cdata, integer, boolean)
	AS IMPLICIT;

CREATE CAST (numeric AS cdata)
	WITH INOUT
	AS ASSIGNMENT;

CREATE OR REPLACE FUNCTION CData(integer, integer, boolean)
	RETURNS cdata
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'int32_to_cdata'
	LANGUAGE C IMMUTABLE STRICT;

CREATE CAST (integer AS cdata)
	WITH FUNCTION CData(integer, integer, boolean)
	AS ASSIGNMENT;

CREATE OR REPLACE FUNCTION Float8(cdata)
	RETURNS double precision
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'cdata_to_float8'
	LANGUAGE C IMMUTABLE STRICT;

CREATE CAST (cdata AS double precision)
	WITH FUNCTION Float8(cdata);
	-- EXPLICIT

CREATE OR REPLACE FUNCTION CData_update(cdata, cdata)
	RETURNS cdata
	AS 'postrr-@POSTRR_MAJOR_VERSION@.@POSTRR_MINOR_VERSION@', 'cdata_update'
	LANGUAGE C IMMUTABLE;

CREATE OR REPLACE FUNCTION PostRR_update(name, name, name, timestamptz, double precision)
	RETURNS cdata
	LANGUAGE plpgsql
	AS $$
DECLARE
	tbl ALIAS FOR $1;
	tscol ALIAS FOR $2;
	vcol ALIAS FOR $3;
	ts ALIAS FOR $4;
	value ALIAS FOR $5;
	ts_str text;
	v_str text;
	update_qry text;
	status integer;
	newts rrtimeslice;
	new cdata;
BEGIN
	tscol  := quote_ident(tscol);
	vcol   := quote_ident(vcol);
	ts_str := quote_literal(ts);
	v_str  := quote_literal(value);

	update_qry = 'CData_update(' || vcol || ', ' || v_str || ')';

	-- XXX: handle race conditions

	LOOP
		BEGIN
			-- removing matching (by sequence no) old entries
			EXECUTE 'DELETE FROM ' || tbl
				|| ' WHERE rrtimeslice_cmp(' || tscol || ', ' || ts_str
				|| ') = -1; '
				|| 'SELECT rrtimeslice_cmp(' || tscol || ', ' || ts_str
				|| ') AS status FROM ' || tbl
				|| ' WHERE ' || tscol || ' = ' || ts_str
				INTO STRICT status;

			EXCEPTION
				WHEN NO_DATA_FOUND THEN
					EXECUTE 'INSERT INTO ' || tbl
						|| ' (' || tscol || ', ' || vcol
						|| ') VALUES (' || ts_str || ', ' || v_str
						|| ') RETURNING ' || tscol || ', ' || vcol
						INTO newts, new;
					-- use strict again; on exception retry?
					RETURN new;
				WHEN TOO_MANY_ROWS THEN
					RAISE EXCEPTION '% is not unique in %.%',
						ts_str, tbl, tscol;
		END;

		IF status = 0 THEN
			-- timestamps match
			EXECUTE 'UPDATE ' || tbl
				|| ' SET ' || vcol || ' = ' || update_qry
				|| ' WHERE ' || tscol || ' = ' || ts_str
				|| ' RETURNING ' || tscol || ', ' || vcol
				INTO STRICT newts, new;
		ELSIF status < 0 THEN
			-- someone else inserted older data in the meantime
			-- => try again
			CONTINUE;
		ELSE
			RAISE EXCEPTION '% is too old in %.%', ts_str, tbl, tscol;
		END IF;
		EXIT;
	END LOOP;
	RETURN new;
END;
$$;

CREATE OR REPLACE FUNCTION PostRR_update(text, timestamptz, double precision)
	RETURNS SETOF cdata
	LANGUAGE plpgsql
	AS $$
DECLARE
	-- $1: rraname
	-- $2: timestamp
	-- $3: value
	adef RECORD;
	new cdata;
BEGIN
	FOR adef IN SELECT tbl, tscol, vcol FROM postrr.rrarchives
			WHERE postrr.rrarchives.rraname = $1 LOOP
		SELECT PostRR_update(adef.tbl, adef.tscol, adef.vcol, $2, $3)
			INTO new;
		RETURN NEXT new;
	END LOOP;
	RETURN;
END;
$$;

-- vim: set tw=78 sw=4 ts=4 noexpandtab :