slony1_funcs.sql.in

-- ----------------------------------------------------------------------
-- slony1_funcs.sql
--
--    Declaration of replication support functions.
--
--	Copyright (c) 2003-2004, PostgreSQL Global Development Group
--	Author: Jan Wieck, Afilias USA INC.
--
-- $Id: slony1_funcs.sql.in,v 1.25 2004-03-13 21:53:21 wieck Exp $
-- ----------------------------------------------------------------------


-- **********************************************************************
-- * C functions in src/backend/slony1_base.c
-- **********************************************************************


-- ----------------------------------------------------------------------
-- FUNCTION createEvent (cluster_name, ev_type [, ev_data [...]])
--
--	Create an sl_event entry
-- ----------------------------------------------------------------------
create function @NAMESPACE@.createEvent (name, text)
	returns bigint
	as '$libdir/slony1_funcs', '_Slony_I_createEvent'
	language C
	called on null input;

create function @NAMESPACE@.createEvent (name, text, text)
	returns bigint
	as '$libdir/slony1_funcs', '_Slony_I_createEvent'
	language C
	called on null input;

create function @NAMESPACE@.createEvent (name, text, text, text)
	returns bigint
	as '$libdir/slony1_funcs', '_Slony_I_createEvent'
	language C
	called on null input;

create function @NAMESPACE@.createEvent (name, text, text, text, text)
	returns bigint
	as '$libdir/slony1_funcs', '_Slony_I_createEvent'
	language C
	called on null input;

create function @NAMESPACE@.createEvent (name, text, text, text, text, text)
	returns bigint
	as '$libdir/slony1_funcs', '_Slony_I_createEvent'
	language C
	called on null input;

create function @NAMESPACE@.createEvent (name, text, text, text, text, text, text)
	returns bigint
	as '$libdir/slony1_funcs', '_Slony_I_createEvent'
	language C
	called on null input;

create function @NAMESPACE@.createEvent (name, text, text, text, text, text, text, text)
	returns bigint
	as '$libdir/slony1_funcs', '_Slony_I_createEvent'
	language C
	called on null input;

create function @NAMESPACE@.createEvent (name, text, text, text, text, text, text, text, text)
	returns bigint
	as '$libdir/slony1_funcs', '_Slony_I_createEvent'
	language C
	called on null input;

create function @NAMESPACE@.createEvent (name, text, text, text, text, text, text, text, text, text)
	returns bigint
	as '$libdir/slony1_funcs', '_Slony_I_createEvent'
	language C
	called on null input;


-- ----------------------------------------------------------------------
-- FUNCTION getLocalNodeId (name)
--
--	
-- ----------------------------------------------------------------------
create function @NAMESPACE@.getLocalNodeId (name) returns int4
    as '$libdir/slony1_funcs', '_Slony_I_getLocalNodeId'
	language C
	security definer;


-- ----------------------------------------------------------------------
-- FUNCTION setSessionRole (name, role)
--
--	
-- ----------------------------------------------------------------------
create function @NAMESPACE@.setSessionRole (name, text) returns text
    as '$libdir/slony1_funcs', '_Slony_I_setSessionRole'
	language C
	security definer;


-- ----------------------------------------------------------------------
-- FUNCTION getSessionRole (name, role)
--
--	
-- ----------------------------------------------------------------------
create function @NAMESPACE@.getSessionRole (name) returns text
    as '$libdir/slony1_funcs', '_Slony_I_getSessionRole'
	language C
	security definer;


-- ----------------------------------------------------------------------
-- FUNCTION logTrigger ()
--
--	
-- ----------------------------------------------------------------------
create function @NAMESPACE@.logTrigger () returns trigger
    as '$libdir/slony1_funcs', '_Slony_I_logTrigger'
	language C
	security definer;


-- **********************************************************************
-- * PL/pgSQL functions for administrative tasks
-- **********************************************************************


-- ----------------------------------------------------------------------
-- FUNCTION initializeLocalNode (no_id, no_comment)
--
--	Initializes a new node.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.initializeLocalNode (int4, text)
returns int4
as '
declare
	p_local_node_id		alias for $1;
	p_comment			alias for $2;
	v_old_node_id		int4;
	v_first_log_no		int4;
	v_event_seq			int8;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Make sure this node is uninitialized or got reset
	-- ----
	select last_value::int4 into v_old_node_id from @NAMESPACE@.sl_local_node_id;
	if v_old_node_id != -1 then
		raise exception ''Slony-I: This node is already initialized'';
	end if;

	-- ----
	-- Set sl_local_node_id to the requested value and add our
	-- own system to sl_node.
	-- ----
	perform setval(''@NAMESPACE@.sl_local_node_id'', p_local_node_id);
	perform @NAMESPACE@.storeNode_int (p_local_node_id, p_comment);
	
	return p_local_node_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION resetLocalNode ()
--
--	Reset the local node to appear uninitialized.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.resetLocalNode ()
returns int4
as '
begin
	-- **** TODO ****
	raise exception ''Slony-I: resetLocalNode() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION storeNode (no_id, no_comment)
--
--	Generate the STORE_NODE event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.storeNode (int4, text)
returns int4
as '
declare
	p_no_id			alias for $1;
	p_no_comment	alias for $2;
begin
	perform @NAMESPACE@.storeNode_int (p_no_id, p_no_comment);
	perform @NAMESPACE@.createEvent(''_@CLUSTERNAME@'', ''STORE_NODE'',
									p_no_id, p_no_comment);

	return p_no_id;
end;
' language plpgsql
	called on null input;


-- ----------------------------------------------------------------------
-- FUNCTION storeNode_int (no_id, no_comment)
--
--	Process the STORE_NODE event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.storeNode_int (int4, text)
returns int4
as '
declare
	p_no_id			alias for $1;
	p_no_comment	alias for $2;
	v_old_row		record;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Check if the node exists
	-- ----
	select * into v_old_row
			from @NAMESPACE@.sl_node
			where no_id = p_no_id
			for update;
	if found then 
		-- ----
		-- Node exists, update the existing row.
		-- ----
		update @NAMESPACE@.sl_node
				set no_comment = p_no_comment
				where no_id = p_no_id;
	else
		-- ----
		-- New node, insert the sl_node row
		-- ----
		insert into @NAMESPACE@.sl_node
				(no_id, no_active, no_comment) values
				(p_no_id, ''f'', p_no_comment);
	end if;

	return p_no_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION enableNode (no_id)
--
--	Generate the ENABLE_NODE event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.enableNode (int4)
returns int4
as '
declare
	p_no_id			alias for $1;
	v_local_node_id	int4;
	v_node_row		record;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Check that we are the node to activate and that we are
	-- currently disabled.
	-- ----
	v_local_node_id := @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'');
	select * into v_node_row
			from @NAMESPACE@.sl_node
			where no_id = p_no_id
			for update;
	if not found then 
		raise exception ''Slony-I: node % not found'', p_no_id;
	end if;
	if v_node_row.no_active then
		raise exception ''Slony-I: node % is already active'', p_no_id;
	end if;

	-- ----
	-- Activate this node and generate the ENABLE_NODE event
	-- ----
	perform @NAMESPACE@.enableNode_int (p_no_id);
	perform @NAMESPACE@.createEvent(''_@CLUSTERNAME@'', ''ENABLE_NODE'',
									p_no_id);

	return p_no_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION enableNode_int (no_id)
--
--	Process the ENABLE_NODE event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.enableNode_int (int4)
returns int4
as '
declare
	p_no_id			alias for $1;
	v_local_node_id	int4;
	v_node_row		record;
	v_sub_row		record;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Check that the node is inactive
	-- ----
	select * into v_node_row
			from @NAMESPACE@.sl_node
			where no_id = p_no_id
			for update;
	if not found then 
		raise exception ''Slony-I: node % not found'', p_no_id;
	end if;
	if v_node_row.no_active then
		return p_no_id;
	end if;

	-- ----
	-- Activate the node and generate sl_confirm status rows for it.
	-- ----
	update @NAMESPACE@.sl_node
			set no_active = ''t''
			where no_id = p_no_id;
	insert into @NAMESPACE@.sl_confirm
			(con_origin, con_received, con_seqno)
			select no_id, p_no_id, 0 from @NAMESPACE@.sl_node
				where no_id != p_no_id
				and no_active;
	insert into @NAMESPACE@.sl_confirm
			(con_origin, con_received, con_seqno)
			select p_no_id, no_id, 0 from @NAMESPACE@.sl_node
				where no_id != p_no_id
				and no_active;

	-- ----
	-- Generate ENABLE_SUBSCRIPTION events for all sets that
	-- origin here and are subscribed by the just enabled node.
	-- ----
	v_local_node_id := @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'');
	for v_sub_row in select SUB.sub_set from
			@NAMESPACE@.sl_set S,
			@NAMESPACE@.sl_subscribe SUB
			where S.set_origin = v_local_node_id
			and S.set_id = SUB.sub_set
			and SUB.sub_receiver = p_no_id
			for update of S
	loop
		perform @NAMESPACE@.enableSubscription (v_sub_row.sub_set, p_no_id);
	end loop;

	return p_no_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION disableNode (no_id)
--
--	Generate the DISABLE_NODE event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.disableNode (int4)
returns int4
as '
declare
	p_no_id			alias for $1;
begin
	-- **** TODO ****
	raise exception ''Slony-I: disableNode() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION disableNode_int (no_id)
--
--	Process the DISABLE_NODE event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.disableNode_int (int4)
returns int4
as '
declare
	p_no_id			alias for $1;
begin
	-- **** TODO ****
	raise exception ''Slony-I: disableNode_int() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION dropNode (no_id)
--
--	Generate the DROP_NODE event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.dropNode (int4)
returns int4
as '
declare
	p_no_id			alias for $1;
begin
	-- **** TODO ****
	raise exception ''Slony-I: dropNode() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION dropNode_int (no_id)
--
--	Process the DROP_NODE event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.dropNode_int (int4)
returns int4
as '
declare
	p_no_id			alias for $1;
begin
	-- **** TODO ****
	raise exception ''Slony-I: dropNode_int() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION storePath (pa_server, pa_client, pa_conninfo, pa_connretry)
--
--	Generate the STORE_PATH event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.storePath (int4, int4, text, int4)
returns int4
as '
declare
	p_pa_server		alias for $1;
	p_pa_client		alias for $2;
	p_pa_conninfo	alias for $3;
	p_pa_connretry	alias for $4;
begin
	perform @NAMESPACE@.storePath_int(p_pa_server, p_pa_client,
			p_pa_conninfo, p_pa_connretry);
	perform @NAMESPACE@.createEvent(''_@CLUSTERNAME@'', ''STORE_PATH'', 
			p_pa_server, p_pa_client, p_pa_conninfo, p_pa_connretry);

	return 0;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION storePath_int (pa_server, pa_client, pa_conninfo, pa_connretry)
--
--	Process the STORE_PATH event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.storePath_int (int4, int4, text, int4)
returns int4
as '
declare
	p_pa_server		alias for $1;
	p_pa_client		alias for $2;
	p_pa_conninfo	alias for $3;
	p_pa_connretry	alias for $4;
	v_dummy			int4;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Check if the path already exists
	-- ----
	select 1 into v_dummy
			from @NAMESPACE@.sl_path
			where pa_server = p_pa_server
			and pa_client = p_pa_client
			for update;
	if found then
		-- ----
		-- Path exists, update pa_conninfo
		-- ----
		update @NAMESPACE@.sl_path
				set pa_conninfo = p_pa_conninfo,
					pa_connretry = p_pa_connretry
				where pa_server = p_pa_server
				and pa_client = p_pa_client;
	else
		-- ----
		-- New path
		--
		-- In case we receive STORE_PATH events before we know
		-- about the nodes involved in this, we generate those nodes
		-- as pending.
		-- ----
		if not exists (select 1 from @NAMESPACE@.sl_node
						where no_id = p_pa_server) then
			perform @NAMESPACE@.storeNode_int (p_pa_server, ''<event pending>'');
		end if;
		if not exists (select 1 from @NAMESPACE@.sl_node
						where no_id = p_pa_client) then
			perform @NAMESPACE@.storeNode_int (p_pa_client, ''<event pending>'');
		end if;
		insert into @NAMESPACE@.sl_path
				(pa_server, pa_client, pa_conninfo, pa_connretry) values
				(p_pa_server, p_pa_client, p_pa_conninfo, p_pa_connretry);
	end if;

	return 0;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION dropPath (pa_server, pa_client)
--
--	Generate the DROP_PATH event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.dropPath (int4, int4)
returns int4
as '
declare
	p_pa_server		alias for $1;
	p_pa_client		alias for $2;
begin
	-- **** TODO ****
	raise exception ''Slony-I: dropPath() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION dropPath_int (pa_server, pa_client)
--
--	Process the DROP_NODE event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.dropPath_int (int4, int4)
returns int4
as '
declare
	p_pa_server		alias for $1;
	p_pa_client		alias for $2;
begin
	-- **** TODO ****
	raise exception ''Slony-I: dropPath_int() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION storeListen (li_origin, li_provider, li_receiver)
--
--	Generate the STORE_LISTEN event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.storeListen (int4, int4, int4)
returns int4
as '
declare
	p_li_origin		alias for $1;
	p_li_provider	alias for $2;
	p_li_receiver	alias for $3;
begin
	perform @NAMESPACE@.storeListen_int (p_li_origin, p_li_provider, p_li_receiver);
	perform @NAMESPACE@.createEvent (''_@CLUSTERNAME@'', ''STORE_LISTEN'',
			p_li_origin, p_li_provider, p_li_receiver);

	return 0;
end;
' language plpgsql
	called on null input;


-- ----------------------------------------------------------------------
-- FUNCTION storeListen_int (li_origin, li_provider, li_receiver)
--
--	Process the STORE_LISTEN event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.storeListen_int (int4, int4, int4)
returns int4
as '
declare
	p_li_origin		alias for $1;
	p_li_provider	alias for $2;
	p_li_receiver	alias for $3;
	v_exists		int4;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	select 1 into v_exists
			from @NAMESPACE@.sl_listen
			where li_origin = p_li_origin
			and li_provider = p_li_provider
			and li_receiver = p_li_receiver;
	if not found then
		-- ----
		-- In case we receive STORE_LISTEN events before we know
		-- about the nodes involved in this, we generate those nodes
		-- as pending.
		-- ----
		if not exists (select 1 from @NAMESPACE@.sl_node
						where no_id = p_li_origin) then
			perform @NAMESPACE@.storeNode_int (p_li_origin, ''<event pending>'');
		end if;
		if not exists (select 1 from @NAMESPACE@.sl_node
						where no_id = p_li_provider) then
			perform @NAMESPACE@.storeNode_int (p_li_provider, ''<event pending>'');
		end if;
		if not exists (select 1 from @NAMESPACE@.sl_node
						where no_id = p_li_receiver) then
			perform @NAMESPACE@.storeNode_int (p_li_receiver, ''<event pending>'');
		end if;

		insert into @NAMESPACE@.sl_listen
				(li_origin, li_provider, li_receiver) values
				(p_li_origin, p_li_provider, p_li_receiver);
	end if;

	return 0;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION dropListen (li_origin, li_provider, li_receiver)
--
--	Generate the DROP_LISTEN event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.dropListen (int4, int4, int4)
returns int4
as '
declare
	p_li_origin		alias for $1;
	p_li_provider	alias for $2;
	p_li_receiver	alias for $3;
	v_retval		int4;
begin
	v_retval := @NAMESPACE@.dropListen_int(p_li_origin, 
			p_li_provider, p_li_receiver);
	
	perform @NAMESPACE@.createEvent (''_@CLUSTERNAME@'', ''DROP_LISTEN'',
			p_li_origin, p_li_provider, p_li_receiver);

	return v_retval;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION dropListen_int (li_origin, li_provider, li_receiver)
--
--	Process the DROP_LISTEN event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.dropListen_int (int4, int4, int4)
returns int4
as '
declare
	p_li_origin		alias for $1;
	p_li_provider	alias for $2;
	p_li_receiver	alias for $3;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	delete from @NAMESPACE@.sl_listen
			where li_origin = p_li_origin
			and li_provider = p_li_provider
			and li_receiver = p_li_receiver;
	if found then
		return 1;
	else
		return 0;
	end if;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION storeSet (set_id, set_comment)
--
--	Generate the STORE_SET event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.storeSet (int4, text)
returns int4
as '
declare
	p_set_id			alias for $1;
	p_set_comment		alias for $2;
	v_local_node_id		int4;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	v_local_node_id := @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'');

	insert into @NAMESPACE@.sl_set
			(set_id, set_origin, set_comment) values
			(p_set_id, v_local_node_id, p_set_comment);

	perform @NAMESPACE@.createEvent(''_@CLUSTERNAME@'', ''STORE_SET'', 
			p_set_id, v_local_node_id, p_set_comment);

	return p_set_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION storeSet_int (set_id, set_origin, set_comment)
--
--	Process the STORE_SET event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.storeSet_int (int4, int4, text)
returns int4
as '
declare
	p_set_id			alias for $1;
	p_set_origin		alias for $2;
	p_set_comment		alias for $3;
	v_dummy				int4;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	select 1 into v_dummy
			from @NAMESPACE@.sl_set
			where set_id = p_set_id
			for update;
	if found then 
		update @NAMESPACE@.sl_set
				set set_comment = p_set_comment
				where set_id = p_set_id;
	else
		if not exists (select 1 from @NAMESPACE@.sl_node
						where no_id = p_set_origin) then
			perform @NAMESPACE@.storeNode_int (p_set_origin, ''<event pending>'');
		end if;
		insert into @NAMESPACE@.sl_set
				(set_id, set_origin, set_comment) values
				(p_set_id, p_set_origin, p_set_comment);
	end if;

	return p_set_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION moveSet (set_id)
--
--	Generate the MOVE_SET event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.moveSet (int4)
returns int4
as '
declare
	p_set_id			alias for $1;
	v_local_node_id		int4;
	v_set_row			record;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Check that the set exists and that it is currently remote
	-- ----
	v_local_node_id := @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'');
	select * into v_set_row from @NAMESPACE@.sl_set
			where set_id = p_set_id
			for update;
	if not found then
		raise exception ''Slony-I: set % not found'', p_set_id;
	end if;
	if v_set_row.set_origin = v_local_node_id then
		raise exception ''Slony-I: set % originates on local node'',
				p_set_id;
	end if;

	-- ----
	-- Take over the set and generate the message
	-- ----
	update @NAMESPACE@.sl_set
			set set_origin = v_local_node_id
			where set_id = p_set_id;
	perform @NAMESPACE@.createEvent(''_@CLUSTERNAME@'', ''MOVE_SET'', 
			p_set_id, v_local_node_id);

	return p_set_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION moveSet_int (set_id, new_set_origin)
--
--	Process the MOVE_SET event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.moveSet_int (int4, int4)
returns int4
as '
declare
	p_set_id			alias for $1;
	p_new_set_origin	alias for $2;
	v_set_row			record;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Check that the set exists.
	-- ----
	select * into v_set_row from @NAMESPACE@.sl_set
			where set_id = p_set_id
			for update;
	if not found then
		raise exception ''Slony-I: set % not found'', p_set_id;
	end if;

	-- ----
	-- Change the origin of the set
	-- ----
	update @NAMESPACE@.sl_set
			set set_origin = p_new_set_origin
			where set_id = p_set_id;

	return p_set_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION dropSet (set_id)
--
--	Generate the DROP_SET event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.dropSet (int4)
returns int4
as '
declare
	p_set_id			alias for $1;
begin
	-- **** TODO ****
	raise exception ''Slony-I: dropSet() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION dropSet_int (set_id)
--
--	Process the DROP_SET event.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.dropSet_int (int4)
returns int4
as '
declare
	p_set_id			alias for $1;
begin
	-- **** TODO ****
	raise exception ''Slony-I: dropSet_int() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION setAddTable (set_id, tab_id, tab_fqname, tab_attkind, tab_comment)
-- ----------------------------------------------------------------------
create function @NAMESPACE@.setAddTable(int4, int4, text, text, text)
returns int4
as '
declare
	p_set_id			alias for $1;
	p_tab_id			alias for $2;
	p_fqname			alias for $3;
	p_tab_attkind		alias for $4;
	p_tab_comment		alias for $5;
	v_set_origin		int4;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Check that we are the origin of the set
	-- ----
	select set_origin into v_set_origin
			from @NAMESPACE@.sl_set
			where set_id = p_set_id;
	if not found then
		raise exception ''Slony-I: setAddTable(): set % not found'', p_set_id;
	end if;
	if v_set_origin != @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'') then
		raise exception ''Slony-I: setAddTable(): set % has remote origin'', p_set_id;
	end if;

	-- ----
	-- Add the table to the set and generate the SET_ADD_TABLE event
	-- ----
	perform @NAMESPACE@.setAddTable_int(p_set_id, p_tab_id, p_fqname,
			p_tab_attkind, p_tab_comment);
	perform @NAMESPACE@.createEvent(''_@CLUSTERNAME@'', ''SET_ADD_TABLE'',
			p_set_id, p_tab_id, p_fqname, p_tab_attkind, p_tab_comment);

	return p_tab_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION setAddTable_int (set_id, tab_id, tab_fqname, tab_attkind, tab_comment)
-- ----------------------------------------------------------------------
create function @NAMESPACE@.setAddTable_int(int4, int4, text, text, text)
returns int4
as '
declare
	p_set_id			alias for $1;
	p_tab_id			alias for $2;
	p_fqname			alias for $3;
	p_tab_attkind		alias for $4;
	p_tab_comment		alias for $5;
	v_local_node_id		int4;
	v_set_origin		int4;
	v_sub_provider		int4;
	v_relkind			char;
	v_tab_reloid		oid;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- For sets with a remote origin, check that we are subscribed 
	-- to that set. Otherwise we ignore the table because it might 
	-- not even exist in our database.
	-- ----
	v_local_node_id := @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'');
	select set_origin into v_set_origin
			from @NAMESPACE@.sl_set
			where set_id = p_set_id;
	if not found then
		raise exception ''Slony-I: setAddTable_int(): set % not found'',
				p_set_id;
	end if;
	if v_set_origin != v_local_node_id then
		select sub_provider into v_sub_provider
				from @NAMESPACE@.sl_subscribe
				where sub_set = p_set_id
				and sub_receiver = @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'');
		if not found then
			return 0;
		end if;
	end if;
	
	-- ----
	-- Get the tables OID and check that it is a real table
	-- ----
	select PGC.oid, PGC.relkind into v_tab_reloid, v_relkind
			from "pg_catalog".pg_class PGC, "pg_catalog".pg_namespace PGN
			where PGC.relnamespace = PGN.oid
			and p_fqname = "pg_catalog".quote_ident(PGN.nspname) ||
					''.'' || "pg_catalog".quote_ident(PGC.relname);
	if not found then
		raise exception ''Slony-I: setAddTable(): table % not found'', 
				p_fqname;
	end if;
	if v_relkind != ''r'' then
		raise exception ''Slony-I: setAddTable(): % is not a regular table'',
				p_fqname;
	end if;

	-- ----
	-- Add the table to sl_table and create the trigger on it.
	-- ----
	insert into @NAMESPACE@.sl_table
			(tab_id, tab_reloid, tab_set, tab_attkind, tab_comment) values
			(p_tab_id, v_tab_reloid, p_set_id, p_tab_attkind, p_tab_comment);
	perform @NAMESPACE@.tableAddTrigger(p_tab_id, p_fqname, p_tab_attkind);

	return p_tab_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION tableAddTrigger (tab_id, tab_fqname, attkind)
-- ----------------------------------------------------------------------
create function @NAMESPACE@.tableAddTrigger (int4, text, text)
returns int4
as '
declare
	p_tab_id			alias for $1;
	p_tab_fqname		alias for $2;
	p_tab_attkind		alias for $3;
	v_tgname			name;
begin
	-- ----
	-- The table must have a primary key
	-- ----
	if not p_tab_attkind ~ ''k'' then
		raise exception ''Slony-I: table % does not have a primary key'',
				p_tab_fqname;
	end if;

	-- ----
	-- Check if the log trigger exists already
	-- ----
	v_tgname := ''_@CLUSTERNAME@_logtrigger_'' || p_tab_id;
	if exists (select true 
			from @NAMESPACE@.sl_table T,
			"pg_catalog".pg_trigger PGT
			where T.tab_id = p_tab_id
			and T.tab_reloid = PGT.tgrelid
			and PGT.tgname = v_tgname)
	then
		return p_tab_id;
	end if;


	-- ----
	-- Define the trigger on the table
	-- ----
	execute ''create trigger "_@CLUSTERNAME@_logtrigger_'' || p_tab_id || ''"
			after insert or update or delete on '' || p_tab_fqname || ''
			for each row execute procedure
			@NAMESPACE@.logTrigger (''''_@CLUSTERNAME@'''', '''''' || p_tab_id || '''''', '''''' || p_tab_attkind || '''''');'';

	return p_tab_id;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION subscribeSet (sub_set, sub_provider, sub_receiver, sub_forward)
-- ----------------------------------------------------------------------
create function @NAMESPACE@.subscribeSet (int4, int4, int4, bool)
returns int4
as '
declare
	p_sub_set			alias for $1;
	p_sub_provider		alias for $2;
	p_sub_receiver		alias for $3;
	p_sub_forward		alias for $4;
	v_set_origin		int4;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Check that this is called on the receiver node
	-- ----
	if p_sub_receiver != @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'') then
		raise exception ''Slony-I: subscribeSet() must be called on receiver'';
	end if;

	-- ----
	-- Check that the origin and provider of the set are remote
	-- ----
	select set_origin into v_set_origin
			from @NAMESPACE@.sl_set
			where set_id = p_sub_set;
	if not found then
		raise exception ''Slony-I: set % not found'', p_sub_set;
	end if;
	if v_set_origin = p_sub_receiver then
		raise exception 
				''Slony-I: set origin and receiver cannot be identical'';
	end if;
	if p_sub_receiver = p_sub_provider then
		raise exception 
				''Slony-I: set provider and receiver cannot be identical'';
	end if;

	-- ----
	-- Call the internal procedure to store the subscription
	-- ----
	perform @NAMESPACE@.subscribeSet_int(p_sub_set, p_sub_provider,
			p_sub_receiver, p_sub_forward);

	-- ----
	-- Create the SUBSCRIBE_SET event
	-- ----
	perform @NAMESPACE@.createEvent(''_@CLUSTERNAME@'', ''SUBSCRIBE_SET'', 
			p_sub_set, p_sub_provider, p_sub_receiver, 
			case p_sub_forward when true then ''t'' else ''f'' end);

	return p_sub_set;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION subscribeSet_int (sub_set, sub_provider, sub_receiver, sub_forward)
-- ----------------------------------------------------------------------
create function @NAMESPACE@.subscribeSet_int (int4, int4, int4, bool)
returns int4
as '
declare
	p_sub_set			alias for $1;
	p_sub_provider		alias for $2;
	p_sub_receiver		alias for $3;
	p_sub_forward		alias for $4;
	v_set_origin		int4;
	v_sub_row			record;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- Provider change is only allowed for active sets
	-- ----
	if p_sub_receiver = @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'') then
		select sub_active into v_sub_row from @NAMESPACE@.sl_subscribe
				where sub_set = p_sub_set
				and sub_receiver = p_sub_receiver;
		if found then
			if not v_sub_row.sub_active then
				raise exception ''Slony-I: set % is not active, cannot change provider'',
						p_sub_set;
			end if;
		end if;
	end if;

	-- ----
	-- Try to change provider and/or forward for an existing subscription
	-- ----
	update @NAMESPACE@.sl_subscribe
			set sub_provider = p_sub_provider,
				sub_forward = p_sub_forward
			where sub_set = p_sub_set
			and sub_receiver = p_sub_receiver;
	if found then
		return p_sub_set;
	end if;

	-- ----
	-- Not found, insert a new one
	-- ----
	insert into @NAMESPACE@.sl_subscribe
			(sub_set, sub_provider, sub_receiver, sub_forward, sub_active)
			values (p_sub_set, p_sub_provider, p_sub_receiver,
				p_sub_forward, false);

	-- ----
	-- If the set origin is here, then enable the subscription
	-- ----
	select set_origin into v_set_origin
			from @NAMESPACE@.sl_set
			where set_id = p_sub_set;
	if not found then
		raise exception ''Slony-I: set % not found'', p_sub_set;
	end if;

	if v_set_origin = @NAMESPACE@.getLocalNodeId(''_@CLUSTERNAME@'') then
		perform @NAMESPACE@.createEvent(''_@CLUSTERNAME@'', ''ENABLE_SUBSCRIPTION'', 
				p_sub_set, p_sub_provider, p_sub_receiver, 
				case p_sub_forward when true then ''t'' else ''f'' end);
		perform @NAMESPACE@.enableSubscription(p_sub_set, p_sub_receiver);
	end if;

	return p_sub_set;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION unsubscribeSet (sub_set, sub_receiver)
-- ----------------------------------------------------------------------
create function @NAMESPACE@.unsubscribeSet (int4, int4)
returns int4
as '
declare
	p_sub_set			alias for $1;
	p_sub_receiver		alias for $2;
begin
	-- **** TODO ****
	raise exception ''Slony-I: unsbscribeSet() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION unsubscribeSet_int (sub_set, sub_receiver)
-- ----------------------------------------------------------------------
create function @NAMESPACE@.unsubscribeSet_int (int4, int4)
returns int4
as '
declare
	p_sub_set			alias for $1;
	p_sub_receiver		alias for $2;
begin
	-- **** TODO ****
	raise exception ''Slony-I: unsubscribeSet_int() not implemented'';
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION enableSubscription (sub_set, sub_receiver)
-- ----------------------------------------------------------------------
create function @NAMESPACE@.enableSubscription (int4, int4)
returns int4
as '
declare
	p_sub_set			alias for $1;
	p_sub_receiver		alias for $2;
begin
	perform @NAMESPACE@.enableSubscription_int (p_sub_set, p_sub_receiver);

	return p_sub_set;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION enableSubscription_int (sub_set, sub_receiver)
-- ----------------------------------------------------------------------
create function @NAMESPACE@.enableSubscription_int (int4, int4)
returns int4
as '
declare
	p_sub_set			alias for $1;
	p_sub_receiver		alias for $2;
begin
	-- ----
	-- Grab the central configuration lock
	-- ----
	lock table @NAMESPACE@.sl_config_lock;

	-- ----
	-- The real work is done in the replication engine. All
	-- we have to do here is remembering that it happened.
	-- ----
	update @NAMESPACE@.sl_subscribe
			set sub_active = ''t''
			where sub_set = p_sub_set
			and sub_receiver = p_sub_receiver;

	return p_sub_set;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION forwardConfirm ()
--
-- ----------------------------------------------------------------------
create function @NAMESPACE@.forwardConfirm (int4, int4, int8, timestamp)
returns int8
as '
declare
	p_con_origin	alias for $1;
	p_con_received	alias for $2;
	p_con_seqno		alias for $3;
	p_con_timestamp	alias for $4;
	v_max_seqno		int8;
begin
	select into v_max_seqno max(con_seqno)
			from @NAMESPACE@.sl_confirm
			where con_origin = p_con_origin
			and con_received = p_con_received;
	if not found or v_max_seqno < p_con_seqno then
		insert into @NAMESPACE@.sl_confirm 
				(con_origin, con_received, con_seqno, con_timestamp)
				values (p_con_origin, p_con_received, p_con_seqno,
					p_con_timestamp);
		notify "_@CLUSTERNAME@_Confirm";
		v_max_seqno = p_con_seqno;
	end if;

	return v_max_seqno;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION cleanupEvent ()
--
-- ----------------------------------------------------------------------
create function @NAMESPACE@.cleanupEvent ()
returns int4
as '
declare
	v_max_row	record;
	v_min_row	record;
	v_max_sync	int8;
begin
	-- ----
	-- First remove all but the oldest confirm row per origin,receiver pair
	-- ----
	for v_max_row in select con_origin, con_received, max(con_seqno) as con_seqno
				from @NAMESPACE@.sl_confirm
				group by con_origin, con_received
	loop
		delete from @NAMESPACE@.sl_confirm
				where con_origin = v_max_row.con_origin
				and con_received = v_max_row.con_received
				and con_seqno < v_max_row.con_seqno;
	end loop;

	-- ----
	-- Then remove all events that are confirmed by all nodes in the
	-- whole cluster up to the last SYNC
	-- ----
	for v_min_row in select con_origin, min(con_seqno) as con_seqno
				from @NAMESPACE@.sl_confirm
				group by con_origin
	loop
		select max(ev_seqno) into v_max_sync
				from @NAMESPACE@.sl_event
				where ev_origin = v_min_row.con_origin
				and ev_seqno <= v_min_row.con_seqno
				and ev_type = ''SYNC'';
		if found then
			delete from @NAMESPACE@.sl_event
					where ev_origin = v_min_row.con_origin
					and ev_seqno < v_max_sync;
			-- ----
			-- Get the full SYNC row and remove all log data that
			-- is older than it. We ignore the fancy xip stuff here
			-- because it does not matter if we leave some rows
			-- for the next run or not. But it matters that we can
			-- use a prepared plan without a horribly expensive NOT IN
			-- clause.
			-- ----
			select * into v_max_row
					from @NAMESPACE@.sl_event
					where ev_origin = v_min_row.con_origin
					and ev_seqno = v_max_sync;
			if found then
				delete from @NAMESPACE@.sl_log_1
						where log_origin = v_max_row.ev_origin
						and   log_xid    < v_max_row.ev_minxid;
				delete from @NAMESPACE@.sl_log_2
						where log_origin = v_max_row.ev_origin
						and   log_xid    < v_max_row.ev_minxid;
			end if;
		end if;
	end loop;

	return 0;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION tableAddKey (tab_fqname)
--
--	If the specified table does not have a column 
--	"_Slony-I_<clustername>_rowID", then add it as a bigint
--	with default nextval('"_<clustername>".sl_rowid_seq').
-- ----------------------------------------------------------------------
create function @NAMESPACE@.tableAddKey(text) returns text
as '
declare
	p_tab_fqname	alias for $1;
	v_attkind		text default '''';
	v_attrow		record;
	v_have_serial	bool default ''f'';
begin
	--
	-- Loop over the attributes of this relation
	-- and add a "v" for every user column, and a "k"
	-- if we find the Slony-I special serial column.
	--
	for v_attrow in select PGA.attnum, PGA.attname
			from "pg_catalog".pg_class PGC,
			    "pg_catalog".pg_namespace PGN,
				"pg_catalog".pg_attribute PGA
			where "pg_catalog".quote_ident(PGN.nspname) || ''.'' ||
			    "pg_catalog".quote_ident(PGC.relname) = p_tab_fqname
				and PGN.oid = PGC.relnamespace
				and PGA.attrelid = PGC.oid
				and not PGA.attisdropped
				and PGA.attnum > 0
			order by attnum
	loop
		if v_attrow.attname = ''_Slony-I_@CLUSTERNAME@_rowID'' then
		    v_attkind := v_attkind || ''k'';
			v_have_serial := ''t'';
		else
			v_attkind := v_attkind || ''v'';
		end if;
	end loop;
	
	--
	-- A table must have at least one attribute, so not finding
	-- anything means the table does not exist.
	--
	if not found then
		raise exception ''Slony-I: table % not found'', p_tab_fqname;
	end if;

	--
	-- If it does not have the special serial column, we
	-- have to add it. This will be only half way done.
	-- The function to add the table to the set must finish
	-- these definitions with NOT NULL and UNIQUE after
	-- updating all existing rows.
	--
	if not v_have_serial then
		execute ''lock table '' || p_tab_fqname ||
			'' in access exclusive mode'';
		execute ''alter table '' || p_tab_fqname ||
			'' add column "_Slony-I_@CLUSTERNAME@_rowID" bigint;'';
		execute ''alter table '' || p_tab_fqname ||
			'' alter column "_Slony-I_@CLUSTERNAME@_rowID" '' ||
			'' set default "pg_catalog".nextval(''''@NAMESPACE@.sl_rowid_seq'''');'';

		v_attkind := v_attkind || ''k'';
	end if;

	--
	-- Return the resulting Slony-I attkind
	--
	return v_attkind;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION determineAttKindSerial (tab_fqname)
--
--	A table was that was specified without a primary key is added
--	to the replication. Assume that tableAddKey() was called before
--	and finish the creation of the serial column. The return an
--	attkind according to that.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.determineAttkindSerial(text)
returns text
as '
declare
	p_tab_fqname	alias for $1;
	v_attkind		text default '''';
	v_attrow		record;
	v_have_serial	bool default ''f'';
begin
	--
	-- Loop over the attributes of this relation
	-- and add a "v" for every user column, and a "k"
	-- if we find the Slony-I special serial column.
	--
	for v_attrow in select PGA.attnum, PGA.attname
			from "pg_catalog".pg_class PGC,
			    "pg_catalog".pg_namespace PGN,
				"pg_catalog".pg_attribute PGA
			where "pg_catalog".quote_ident(PGN.nspname) || ''.'' ||
			    "pg_catalog".quote_ident(PGC.relname) = p_tab_fqname
				and PGN.oid = PGC.relnamespace
				and PGA.attrelid = PGC.oid
				and not PGA.attisdropped
				and PGA.attnum > 0
			order by attnum
	loop
		if v_attrow.attname = ''_Slony-I_@CLUSTERNAME@_rowID'' then
		    v_attkind := v_attkind || ''k'';
			v_have_serial := ''t'';
		else
			v_attkind := v_attkind || ''v'';
		end if;
	end loop;
	
	--
	-- A table must have at least one attribute, so not finding
	-- anything means the table does not exist.
	--
	if not found then
		raise exception ''Slony-I: table % not found'', p_tab_fqname;
	end if;

	--
	-- If it does not have the special serial column, we
	-- should not have been called in the first place.
	--
	if not v_have_serial then
		raise exception ''Slony-I: table % does not have the serial key'',
				p_tab_fqname;
	end if;

	execute ''update '' || p_tab_fqname ||
		'' set "_Slony-I_@CLUSTERNAME@_rowID" ='' ||
		'' "pg_catalog".nextval(''''@NAMESPACE@.sl_rowid_seq'''');'';
	execute ''alter table '' || p_tab_fqname ||
		'' add unique ("_Slony-I_@CLUSTERNAME@_rowID");'';
	execute ''alter table '' || p_tab_fqname ||
		'' alter column "_Slony-I_@CLUSTERNAME@_rowID" '' ||
		'' set not null;'';

	--
	-- Return the resulting Slony-I attkind
	--
	return v_attkind;
end;
' language plpgsql;


-- ----------------------------------------------------------------------
-- FUNCTION tableHasSerialKey (tab_fqname)
--
--	Checks if a table has our special serial key column that is
--	used if the table has no natural unique constraint.
-- ----------------------------------------------------------------------
create function @NAMESPACE@.tableHasSerialKey(text) 
returns bool
as '
declare
	p_tab_fqname	alias for $1;
	v_attnum		int2;
begin
	select PGA.attnum into v_attnum
			from "pg_catalog".pg_class PGC,
				"pg_catalog".pg_namespace PGN,
				"pg_catalog".pg_attribute PGA
			where "pg_catalog".quote_ident(PGN.nspname) || ''.'' ||
				"pg_catalog".quote_ident(PGC.relname) = p_tab_fqname
				and PGC.relnamespace = PGN.oid
				and PGA.attrelid = PGC.oid
				and PGA.attname = ''_Slony-I_@CLUSTERNAME@_rowID''
				and not PGA.attisdropped;
	return found;
end;
' language plpgsql;