MAN1_SFX = scriptmgr.1 skytools_upgrade.1 queue_mover.1 queue_splitter.1
MAN1 = qadmin.1 pgqd.1 walmgr3.1 londiste3.1
-FQHTML = $(addprefix html/, $(HTMLS))
+FQHTML = $(addprefix html/doc/, $(HTMLS))
FQMAN1 = $(addprefix man/, $(MAN1))
FQMAN1_SFX = $(addprefix man/, $(MAN1_SFX))
FQMAN5 = $(addprefix man/, $(MAN5))
# londiste.syncer londiste.table_copy
apidoc:
- rm -rf api
- mkdir -p api
- cd ../python && $(EPYDOC) $(EPYARGS) -o ../doc/api $(PY_PKGS)
- cd ../sql/pgq && rm -rf docs/html && $(MAKE) dox
- cd ../sql/pgq_coop && rm -rf docs/html && $(MAKE) dox
- cd ../sql/pgq_node && rm -rf docs/html && $(MAKE) dox
- cd ../sql/londiste && rm -rf docs/html && $(MAKE) dox
+ rm -rf html/api
+ mkdir -p html/api
+ cd ../python && $(EPYDOC) $(EPYARGS) -o ../doc/html/api $(PY_PKGS)
+ cd ../sql/pgq && rm -rf docs/html && $(MAKE) dox && cp -rp docs/html ../../doc/html/pgq
+ cd ../sql/pgq_coop && rm -rf docs/html && $(MAKE) dox && cp -rp docs/html ../../doc/html/pgq_coop
+ cd ../sql/pgq_node && rm -rf docs/html && $(MAKE) dox && cp -rp docs/html ../../doc/html/pgq_node
+ cd ../sql/londiste && rm -rf docs/html && $(MAKE) dox && cp -rp docs/html ../../doc/html/londiste
+ cd ../sql/pgq_ext && rm -rf docs/html && $(MAKE) dox && cp -rp docs/html ../../doc/html/pgq_ext
apiupload: apidoc
-rsync -rtlz api/* $(web)/api
-rsync -rtlz ../sql/londiste/docs/html/* $(web)/londiste/
clean:
- rm -rf api *.xml
+ rm -rf html *.xml
distclean: clean
rm -rf ../sql/pgq/docs/pgq
ifneq ($(XMLTO),no)
man/%.xml: %.txt $(COMMON)
@mkdir -p man
- $(ASCIIDOC) -b docbook -d manpage `$(GETATTRS) $<` -o - $< \
- | $(FIXMAN) > $@
+ $(ASCIIDOC) -b docbook -d manpage `$(GETATTRS) $<` -o $@ $<
man/%.1: man/%.xml
@mkdir -p man
$(XMLTO) -o man man $<
endif
-html/%.html: %.txt $(COMMON)
- @mkdir -p html
+html/doc/%.html: %.txt $(COMMON)
+ @mkdir -p html/doc
LANG=C cat $< \
| sed -e '/^include/b' -e 's,\([A-Za-z.0-9]*\)[.]txt,link:\1.html[],g' \
| $(ASCIIDOC) $(AFLAGS) -a toc `$(GETATTRS) $<` -o - - \
| sed -e "/^[/][*] Workarounds/r ../misc/extra.css" \
> $@
-html/README.html: ../README
- @mkdir -p html
+html/doc/README.html: ../README
+ @mkdir -p html/doc
cat $< \
| sed -e 's,doc/\([!-~]*\)[.]txt,link:\1.html[],g' \
-e 's,http:[!-~]*,&[],g' \
| sed -e "/^[/][*] Workarounds/r ../misc/extra.css" \
> $@
-html/INSTALL.html: ../INSTALL
+html/doc/INSTALL.html: ../INSTALL
@mkdir -p html
$(ASCIIDOC) $(AFLAGS) -o - $< \
| sed -e "/^[/][*] Workarounds/r ../misc/extra.css" \
-Format: 1.4
+Format: 1.52
# This is the Natural Docs languages file for this project. If you change
# anything here, it will apply to THIS PROJECT ONLY. If you'd like to change
-Format: 1.4
+Format: 1.52
Title: PgQ
-Format: 1.4
+Format: 1.52
# This is the Natural Docs topics file for this project. If you change anything
# here, it will apply to THIS PROJECT ONLY. If you'd like to change something
-- i_retry_time - Time when the event should be put back into queue
--
-- Returns:
--- nothing
+-- number of events inserted
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- pgq.retry_queue
-- ----------------------------------------------------------------------
declare
_retry timestamptz;
-- Returns:
-- 0 - queue already exists
-- 1 - queue created
+-- Calls:
+-- pgq.grant_perms(i_queue_name);
+-- pgq.ticker(i_queue_name);
+-- pgq.tune_storage(i_queue_name);
+-- Tables directly manipulated:
+-- insert - pgq.queue
+-- create - pgq.event_N () inherits (pgq.event_template)
+-- create - pgq.event_N_0 .. pgq.event_N_M () inherits (pgq.event_N)
-- ----------------------------------------------------------------------
declare
tblpfx text;
-- Event can be added to it without going via functions,
-- e.g. by COPY.
--
+-- If queue is disabled and GUC session_replication_role <> 'replica'
+-- then raises exception.
+--
+-- or expressed in a different way - an even table of a disabled queue
+-- is returned only on replica
+--
-- Note:
-- The result is valid only during current transaction.
--
--
-- Parameters:
-- x_queue_name - queue name
--- x_force - ignore consumers
+-- x_force - ignore (drop) existing consumers
+-- Returns:
+-- 1 - success
+-- Calls:
+-- pgq.unregister_consumer(queue_name, consumer_name)
+-- perform pgq.ticker(i_queue_name);
+-- perform pgq.tune_storage(i_queue_name);
+-- Tables directly manipulated:
+-- delete - pgq.queue
+-- drop - pgq.event_N (), pgq.event_N_0 .. pgq.event_N_M
-- ----------------------------------------------------------------------
declare
tblname text;
x_retry_time timestamptz)
returns integer as $$
-- ----------------------------------------------------------------------
--- Function: pgq.event_retry(3)
+-- Function: pgq.event_retry(3a)
--
-- Put the event into retry queue, to be processed again later.
--
-- x_retry_time - Time when the event should be put back into queue
--
-- Returns:
--- nothing
+-- 1 - success
+-- 0 - event already in retry queue
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- insert - pgq.retry_queue
-- ----------------------------------------------------------------------
begin
insert into pgq.retry_queue (ev_retry_after, ev_queue,
x_retry_seconds integer)
returns integer as $$
-- ----------------------------------------------------------------------
--- Function: pgq.event_retry(3)
+-- Function: pgq.event_retry(3b)
--
-- Put the event into retry queue, to be processed later again.
--
-- x_retry_seconds - Time when the event should be put back into queue
--
-- Returns:
--- nothing
+-- 1 - success
+-- 0 - event already in retry queue
+-- Calls:
+-- pgq.event_retry(3a)
+-- Tables directly manipulated:
+-- None
-- ----------------------------------------------------------------------
declare
new_retry timestamptz;
-- x_batch_id - id of batch.
--
-- Returns:
--- If batch 1 if batch was found, 0 otherwise.
+-- 1 if batch was found, 0 otherwise.
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- update - pgq.subscription
-- ----------------------------------------------------------------------
begin
update pgq.subscription
--
-- Returns:
-- List of events.
+-- Calls:
+-- pgq.batch_event_sql(i_batch_id) - internal function which generates SQL optimised specially for getting events in this batch
-- ----------------------------------------------------------------------
declare
_cname text;
--
-- Returns:
-- List of events.
+-- Calls:
+-- pgq.get_batch_cursor(4)
-- ----------------------------------------------------------------------
begin
for ev_id, ev_time, ev_txid, ev_retry, ev_type, ev_data,
-- Parameters:
-- x_batch_id - id of a active batch.
--
--- Returns:
--- Info
+-- Returns: ??? pls check
+-- queue_name - which queue this batch came from
+-- consumer_name - batch processed by
+-- batch_start - start time of batch
+-- batch_end - end time of batch
+-- prev_tick_id - start tick for this batch
+-- tick_id - end tick for this batch
+-- lag - now() - tick_id.time
+-- seq_start - start event id for batch
+-- seq_end - end event id for batch
-- ----------------------------------------------------------------------
begin
select q.queue_name, c.co_name,
--
-- Returns:
-- List of pgq.ret_queue_info records.
+-- queue_name - queue name
+-- queue_ntables - number of tables in this queue
+-- queue_cur_table - ???
+-- queue_rotation_period - how often the event_N_M tables in this queue are rotated
+-- queue_switch_time - ??? when was this queue last rotated
+-- queue_external_ticker - ???
+-- queue_ticker_paused - ??? is ticker paused in this queue
+-- queue_ticker_max_count - max number of events before a tick is issued
+-- queue_ticker_max_lag - maks time without a tick
+-- queue_ticker_idle_period - how often the ticker should check this queue
+-- ticker_lag - time from last tick
+-- ev_per_sec - how many events per second this queue serves
+-- ev_new - ???
+-- last_tick_id - last tick id for this queue
+--
-- ----------------------------------------------------------------------
begin
for queue_name, queue_ntables, queue_cur_table, queue_rotation_period,
--
-- Returns:
-- One pgq.ret_queue_info record.
+-- contente same as forpgq.get_queue_info()
-- ----------------------------------------------------------------------
declare
_ticker_lag interval;
--
-- Returns:
-- Event ID
+-- Calls:
+-- pgq.insert_event(7)
-- ----------------------------------------------------------------------
begin
return pgq.insert_event(queue_name, ev_type, ev_data, null, null, null, null);
--
-- Returns:
-- Event ID
+-- Calls:
+-- pgq.insert_event_raw(11)
+-- Tables directly manipulated:
+-- insert - pgq.insert_event_raw(11), a C function, inserts into current event_N_M table
-- ----------------------------------------------------------------------
begin
return pgq.insert_event_raw(queue_name, null, now(), null, null,
-- prev_tick_id - Start tick id.
-- prev_tick_time - Start tick time.
-- prev_tick_event_seq - value from event id sequence at the time tick was issued.
+-- Calls:
+-- pgq.next_batch_custom(5)
+-- Tables directly manipulated:
+-- None
-- ----------------------------------------------------------------------
begin
select f.batch_id, f.cur_tick_id, f.prev_tick_id,
-- prev_tick_id - Start tick id.
-- prev_tick_time - Start tick time.
-- prev_tick_event_seq - value from event id sequence at the time tick was issued.
+-- Calls:
+-- pgq.insert_event_raw(11)
+-- Tables directly manipulated:
+-- update - pgq.subscription
-- ----------------------------------------------------------------------
declare
errmsg text;
-- Returns:
-- 0 - if already registered
-- 1 - if new registration
+-- Calls:
+-- pgq.register_consumer_at(3)
+-- Tables directly manipulated:
+-- None
-- ----------------------------------------------------------------------
begin
return pgq.register_consumer_at(x_queue_name, x_consumer_id, NULL);
--
-- Returns:
-- 0/1 whether consumer has already registered.
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- update/insert - pgq.subscription
-- ----------------------------------------------------------------------
declare
tmp text;
--
-- Returns:
-- 0 if event was already in queue, 1 otherwise.
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- update - pgq.queue
-- ----------------------------------------------------------------------
declare
v_param_name text;
-- x_consumer_name - Name of the consumer
--
-- Returns:
--- nothing
+-- number of (sub)consumers unregistered
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- delete - pgq.retry_queue
+-- delete - pgq.subscription
-- ----------------------------------------------------------------------
declare
x_sub_id integer;
+-- ----------------------------------------------------------------------
-- Section: Public Functions
+--
+-- The queue is used by a client in the following steps
+--
+-- 1. Register the client (a queue consumer)
+--
+-- pgq.register_consumer(queue_name, consumer_id)
+--
+-- 2. run a loop createing, consuming and closing batches
+--
+-- 2a. pgq.get_batch_events(batch_id int8) - returns an int8 batch handle
+--
+-- 2b. pgq.get_batch_events(batch_id int8) - returns a set of events for current batch
+--
+-- the event structure is :(ev_id int8, ev_time timestamptz, ev_txid int8, ev_retry
+-- int4, ev_type text, ev_data text, ev_extra1, ev_extra2, ev_extra3, ev_extra4)
+--
+-- 2c. if any of the events need to be tagged as failed, use a the function
+--
+-- pgq.event_failed(batch_id int8, event_id int8, reason text)
+--
+-- 2d. if you want the event to be re-inserted in the main queue afrer N seconds, use
+--
+-- pgq.event_retry(batch_id int8, event_id int8, retry_seconds int4)
+--
+-- 2e. To finish processing and release the batch, use
+--
+-- pgq.finish_batch(batch_id int8)
+--
+-- Until this is not done, the consumer will get same batch again.
+--
+-- After calling finish_batch consumer cannot do any operations with events
+-- of that batch. All operations must be done before.
+--
+-- -- ----------------------------------------------------------------------
+
-- Group: Queue creation
-- pgq.event_* - Data tables
-- pgq.retry_queue - Events to be retried later
--
--- Its basically generalized and simplified Slony-I structure:
--- sl_node - pgq.consumer
--- sl_set - pgq.queue
--- sl_subscriber + sl_confirm - pgq.subscription
--- sl_event - pgq.tick
--- sl_setsync - pgq_ext.completed_*
--- sl_log_* - slony1 has per-cluster data tables,
--- pgq has per-queue data tables.
+--
+-- Standard triggers store events in the pgq.event_* data tables
+-- There is one top event table pgq.event_<queue_id> for each queue
+-- inherited from pgq.event_template wuith three tables for actual data
+-- pgq.event_<queue_id>_0 to pgq.event_<queue_id>_2.
+--
+-- The active table is rotated at interval, so that if all the consubers
+-- have passed some poin the oldes one can be emptied using TRUNCATE command
+-- for efficiency
+--
+--
-- ----------------------------------------------------------------------
set client_min_messages = 'warning';
--
-- Returns:
-- 1 if success (batch was found), 0 otherwise
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- update - pgq.subscription
-- ----------------------------------------------------------------------
begin
-- we are dealing with subconsumer, so nullify all tick info
-- i_queue_name - Name of the queue
-- i_consumer_name - Name of the consumer
-- i_subconsumer_name - Name of the subconsumer
+--
+-- Calls:
+-- pgq.register_consumer(i_queue_name, i_consumer_name)
+-- pgq.register_consumer(i_queue_name, _subcon_name);
+--
+-- Tables directly manipulated:
+-- update - pgq.subscription
+--
-- ----------------------------------------------------------------------
begin
return pgq_coop.next_batch_custom(i_queue_name, i_consumer_name, i_subconsumer_name, NULL, NULL, NULL, NULL);
-- Result NULL means nothing to work with, for a moment
--
-- Parameters:
--- i_queue_name - Name of the queue
--- i_consumer_name - Name of the consumer
--- i_subconsumer_name - Name of the subconsumer
+-- i_queue_name - Name of the queue
+-- i_consumer_name - Name of the consumer
+-- i_subconsumer_name - Name of the subconsumer
-- i_min_lag - Consumer wants events older than that
-- i_min_count - Consumer wants batch to contain at least this many events
-- i_min_interval - Consumer wants batch to cover at least this much time
-- i_dead_interval - Take over other subconsumer batch if inactive
+-- Calls:
+-- pgq.register_subconsumer(i_queue_name, i_consumer_name, i_subconsumer_name)
+-- pgq.next_batch_custom(i_queue_name, i_consumer_name, i_min_lag, i_min_count, i_min_interval)
+-- Tables directly manipulated:
+-- update - pgq.subscription
-- ----------------------------------------------------------------------
declare
_queue_id integer;
-- ----------------------------------------------------------------------
-- Function: pgq_coop.register_subconsumer(3)
--
--- Subscribe a subconsumer on a queue.
+-- Subscribe a subconsumer on a queue.
--
-- Subconsumer will be registered as another consumer on queue,
-- whose name will be i_consumer_name and i_subconsumer_name
-- combined.
--
-- Returns:
--- 0 - if already registered
--- 1 - if this is a new registration
+-- 0 - if already registered
+-- 1 - if this is a new registration
+--
+-- Calls:
+-- pgq.register_consumer(i_queue_name, i_consumer_name)
+-- pgq.register_consumer(i_queue_name, _subcon_name);
+--
+-- Tables directly manipulated:
+-- update - pgq.subscription
+--
-- ----------------------------------------------------------------------
declare
_subcon_name text; -- consumer + subconsumer
-- 1 - Close the batch, ignoring the events.
--
-- Returns:
--- nothing?
+-- 0 - no consumer found
+-- 1 - consumer found and unregistered
+--
+-- Tables directly manipulated:
+-- delete - pgq.subscription
+--
-- ----------------------------------------------------------------------
declare
_current_batch bigint;
+
+-- ----------------------------------------------------------------------
-- Section: Functions
+--
+-- Overview:
+--
+-- The usual flow of a cooperative consumer is to
+--
+-- 1. register itself as a subconsumer for a queue:
+-- pgq_coop.register_subconsumer()
+--
+-- And the run a loop doing
+--
+-- 2A. pgq_coop.next_batch ()
+--
+-- 2B. pgq_coop.finish_batch()
+--
+-- Once the cooperative (or sub-)consuber is done, it should unregister
+-- itself before exiting
+--
+-- 3. pgq_coop.unregister_subconsumer()
+--
+--
+-- ----------------------------------------------------------------------
-- Group: Subconsumer registration
\i functions/pgq_coop.register_subconsumer.sql
REGRESS_OPTS = --load-language=plpgsql
CATSQL = ../../scripts/catsql.py
+NDOC = NaturalDocs
+NDOCARGS = -r -o html docs/html -p docs -i docs/sql
PG_CONFIG = pg_config
PGXS = $(shell $(PG_CONFIG) --pgxs)
ack:
cp results/* expected/
+cleandox:
+ rm -rf docs/html docs/Data docs/sql
+
+dox: cleandox $(SRCS)
+ mkdir -p docs/html
+ mkdir -p docs/sql
+ $(CATSQL) --ndoc structure/tables.sql > docs/sql/schema.sql
+ $(CATSQL) --ndoc structure/upgrade.sql > docs/sql/functions.sql
+ $(NDOC) $(NDOCARGS)
+
create or replace function pgq_ext.get_last_tick(a_consumer text, a_subconsumer text)
returns int8 as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.get_last_tick(2)
+--
+-- Gets last completed tick for this consumer
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_subconsumer - subconsumer name
+--
+-- Returns:
+-- tick_id - last completed tick
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- None
+-- ----------------------------------------------------------------------
declare
res int8;
begin
create or replace function pgq_ext.get_last_tick(a_consumer text)
returns int8 as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.get_last_tick(1)
+--
+-- Gets last completed tick for this consumer
+--
+-- Parameters:
+-- a_consumer - consumer name
+--
+-- Returns:
+-- tick_id - last completed tick
+-- Calls:
+-- pgq_ext.get_last_tick(2)
+-- Tables directly manipulated:
+-- None
+-- ----------------------------------------------------------------------
begin
return pgq_ext.get_last_tick(a_consumer, '');
end;
a_subconsumer text,
a_batch_id bigint)
returns boolean as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.is_batch_done(3)
+--
+-- Checks if a certain consumer and subconsumer have completed the batch
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_subconsumer - subconsumer name
+-- a_batch_id - a batch id
+--
+-- Returns:
+-- true if batch is done, else false
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- None
+-- ----------------------------------------------------------------------
declare
res boolean;
begin
a_consumer text,
a_batch_id bigint)
returns boolean as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.is_batch_done(2)
+--
+-- Checks if a certain consumer has completed the batch
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_batch_id - a batch id
+--
+-- Returns:
+-- true if batch is done, else false
+-- Calls:
+-- pgq_ext.is_batch_done(3)
+-- Tables directly manipulated:
+-- None
+-- ----------------------------------------------------------------------
begin
return pgq_ext.is_batch_done(a_consumer, '', a_batch_id);
end;
a_batch_id bigint,
a_event_id bigint)
returns boolean as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.is_event_done(4)
+--
+-- Checks if a certain consumer and subconsumer have "done" and event
+-- in a batch
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_subconsumer - subconsumer name
+-- a_batch_id - a batch id
+-- a_event_id - an event id
+--
+-- Returns:
+-- true if event is done, else false
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- None
+-- ----------------------------------------------------------------------
declare
res bigint;
begin
a_batch_id bigint,
a_event_id bigint)
returns boolean as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.is_event_done(3)
+--
+-- Checks if a certain consumer has "done" and event
+-- in a batch
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_batch_id - a batch id
+-- a_event_id - an event id
+--
+-- Returns:
+-- true if event is done, else false
+-- Calls:
+-- Nonpgq_ext.is_event_done(4)
+-- Tables directly manipulated:
+-- None
+-- ----------------------------------------------------------------------
begin
return pgq_ext.is_event_done(a_consumer, '', a_batch_id, a_event_id);
end;
a_subconsumer text,
a_batch_id bigint)
returns boolean as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.set_batch_done(3)
+--
+-- Marks a batch as "done" for certain consumer and subconsumer
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_subconsumer - subconsumer name
+-- a_batch_id - a batch id
+--
+-- Returns:
+-- false if it already was done
+-- true for successfully marking it as done
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- update - pgq_ext.completed_batch
+-- ----------------------------------------------------------------------
begin
if pgq_ext.is_batch_done(a_consumer, a_subconsumer, a_batch_id) then
return false;
a_consumer text,
a_batch_id bigint)
returns boolean as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.set_batch_done(3)
+--
+-- Marks a batch as "done" for certain consumer
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_batch_id - a batch id
+--
+-- Returns:
+-- false if it already was done
+-- true for successfully marking it as done
+-- Calls:
+-- pgq_ext.set_batch_done(3)
+-- Tables directly manipulated:
+-- None
+-- ----------------------------------------------------------------------
begin
return pgq_ext.set_batch_done(a_consumer, '', a_batch_id);
end;
a_batch_id bigint,
a_event_id bigint)
returns boolean as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.set_event_done(4)
+--
+-- Marks and event done in a batch for a certain consumer and subconsumer
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_subconsumer - subconsumer name
+-- a_batch_id - a batch id
+-- a_event_id - an event id
+--
+-- Returns:
+-- false if already done
+-- true on success
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- insert - pgq_ext.partial_batch
+-- delete - pgq_ext.completed_event
+-- update - pgq_ext.partial_batch
+-- insert - pgq_ext.completed_event
+-- ----------------------------------------------------------------------
declare
old_batch bigint;
begin
a_batch_id bigint,
a_event_id bigint)
returns boolean as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.set_event_done(3)
+--
+-- Marks and event done in a batch for a certain consumer and subconsumer
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_batch_id - a batch id
+-- a_event_id - an event id
+--
+-- Returns:
+-- false if already done
+-- true on success
+-- Calls:
+-- pgq_ext.set_event_done(4)
+-- Tables directly manipulated:
+-- None
+-- ----------------------------------------------------------------------
begin
return pgq_ext.set_event_done(a_consumer, '', a_batch_id, a_event_id);
end;
a_subconsumer text,
a_tick_id bigint)
returns integer as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.set_last_tick(3)
+--
+-- records last completed tick for consumer,
+-- removes row if a_tick_id is NULL
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_subconsumer - subconsumer name
+-- a_tick_id - a tick id
+--
+-- Returns:
+-- 1
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- delete - pgq_ext.completed_tick
+-- update - pgq_ext.completed_tick
+-- insert - pgq_ext.completed_tick
+-- ----------------------------------------------------------------------
begin
if a_tick_id is null then
delete from pgq_ext.completed_tick
a_consumer text,
a_tick_id bigint)
returns integer as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.set_last_tick(2)
+--
+-- records last completed tick for consumer,
+-- removes row if a_tick_id is NULL
+--
+-- Parameters:
+-- a_consumer - consumer name
+-- a_tick_id - a tick id
+--
+-- Returns:
+-- 1
+-- Calls:
+-- pgq_ext.set_last_tick(2)
+-- Tables directly manipulated:
+-- None
+-- ----------------------------------------------------------------------
begin
return pgq_ext.set_last_tick(a_consumer, '', a_tick_id);
end;
create or replace function pgq_ext.upgrade_schema()
returns int4 as $$
-- updates table structure if necessary
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.upgrade_schema()
+--
+-- Upgrades tables to have column subconsumer_id
+--
+-- Parameters:
+-- None
+--
+-- Returns:
+-- number of tables updated
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- alter - pgq_ext.completed_batch
+-- alter - pgq_ext.completed_tick
+-- alter - pgq_ext.partial_batch
+-- alter - pgq_ext.completed_event
+-- ----------------------------------------------------------------------
declare
cnt int4 = 0;
tbl text;
create or replace function pgq_ext.version()
returns text as $$
+-- ----------------------------------------------------------------------
+-- Function: pgq_ext.version(0)
+--
+-- Returns version string for pgq_ext. ATM its SkyTools version
+-- with suffix that is only bumped when pgq_ext database code changes.
+-- ----------------------------------------------------------------------
begin
return '3.0.0.3';
end;
+-- ----------------------------------------------------------------------
+-- Section: Tables
+--
+-- The pgq_ext schema exists to help in making sure that allenents get
+-- processed and they get processed only once
+--
+-- Simple quidelines for avoiding duplicate events:
+--
+-- It is pretty burdensome to check if event is already processed,
+-- especially on bulk data moving. Here's a way how checking
+-- individual event checks can be avoided by tracking processing of batches.
+--
+-- First, consumer must guarantee that it processes all events in one tx.
+--
+-- Consumer itself can tag events for retry, but then
+-- it must be able to handle them later.
+--
+-- Simple case: Only one db:
+--
+-- If the PgQ queue and event data handling happen in same database,
+-- the consumer must simply call pgq.finish_batch() inside
+-- the event-processing transaction.
+--
+-- Several databases:
+--
+-- If the event processing happens in different database, the consumer
+-- must store the batch_id into destination database, inside the same
+-- transaction as the event processing happens.
+--
+-- * Only after committing it, consumer can call pgq.finish_batch()
+-- in queue database and commit that.
+--
+-- * As the batches come in sequence, there's no need to remember
+-- full log of batch_id's, it's enough to keep the latest batch_id.
+--
+-- * Then at the start of every batch, consumer can check if the batch_id already
+-- exists in destination database, and if it does, then just tag batch done,
+-- without processing.
+--
+-- With this, there's no need for consumer to check for already processed
+-- events.
+--
+-- Note:
+--
+-- This assumes the event processing is transactional and failures
+-- will be rollbacked. If event processing includes communication with
+-- world outside database, eg. sending email, such handling won't work.
+--
+-- ----------------------------------------------------------------------
set client_min_messages = 'warning';
set default_with_oids = 'off';
--
--- batch tracking
+-- Table: pgq_ext.completed_tick
+--
+-- Used for tracking last completed batch tracking
+-- via tick_id.
+--
+create table pgq_ext.completed_tick (
+ consumer_id text not null,
+ subconsumer_id text not null,
+ last_tick_id bigint not null,
+
+ primary key (consumer_id, subconsumer_id)
+);
+
+--
+-- Table: pgq_ext.completed_batch
+--
+-- Used for tracking last completed batch tracking
--
create table pgq_ext.completed_batch (
consumer_id text not null,
- subconsumer_id text not null default '',
+ subconsumer_id text not null,
last_batch_id bigint not null,
primary key (consumer_id, subconsumer_id)
--
--- event tracking
+-- Table: pgq_ext.completed_event
+--
+-- Stored completed event in current partial batch.
--
create table pgq_ext.completed_event (
consumer_id text not null,
- subconsumer_id text not null default '',
+ subconsumer_id text not null,
batch_id bigint not null,
event_id bigint not null,
primary key (consumer_id, subconsumer_id, batch_id, event_id)
);
-create table pgq_ext.partial_batch (
- consumer_id text not null,
- subconsumer_id text not null default '',
- cur_batch_id bigint not null,
-
- primary key (consumer_id, subconsumer_id)
-);
-
--
--- tick tracking for SerialConsumer()
--- no access functions provided here
+-- Table: pgq_ext.partial_batch
--
-create table pgq_ext.completed_tick (
+-- Stored current in-progress batch
+--
+create table pgq_ext.partial_batch (
consumer_id text not null,
- subconsumer_id text not null default '',
- last_tick_id bigint not null,
+ subconsumer_id text not null,
+ cur_batch_id bigint not null,
primary key (consumer_id, subconsumer_id)
);
+--
+-- Section: Functions
+--
+
\i functions/pgq_ext.upgrade_schema.sql
select pgq_ext.upgrade_schema();
+-- Group: track batches via batch id
\i functions/pgq_ext.is_batch_done.sql
\i functions/pgq_ext.set_batch_done.sql
-\i functions/pgq_ext.is_event_done.sql
-\i functions/pgq_ext.set_event_done.sql
-
+-- Group: track batches via tick id
\i functions/pgq_ext.get_last_tick.sql
\i functions/pgq_ext.set_last_tick.sql
+
+-- Group: Track events separately
+\i functions/pgq_ext.is_event_done.sql
+\i functions/pgq_ext.set_event_done.sql
+
+-- Group: Schema info
\i functions/pgq_ext.version.sql
-- i_queue_name - queue name
-- i_consumer_name - consumer name
-- i_new_provider - node name for new provider
+-- Returns:
+-- ret_code - error code
+-- 200 - ok
+-- 404 - no such consumer or new node
+-- ret_note - description
-- ----------------------------------------------------------------------
begin
perform 1 from pgq_node.node_location
-- i_combined_queue - merge-leaf: target queue
--
-- Returns:
+-- 200 - Ok
-- 401 - node already initialized
+-- ???? - maybe we coud use more error codes ?
--
-- Node Types:
-- root - master node
-- branch - subscriber node that can be provider to others
-- leaf - subscriber node that cannot be provider to others
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- None
-- ----------------------------------------------------------------------
declare
_wm_consumer text;
--
-- Multi-step root demotion to branch.
--
--- Step 1: disable writing to queue.
--- Step 2: wait until writers go away, do tick.
--- Step 3: change type, register.
+-- Must be be called for each step in sequence:
+--
+-- Step 1 - disable writing to queue.
+-- Step 2 - wait until writers go away, do tick.
+-- Step 3 - change type, register.
--
-- Parameters:
-- i_queue_name - queue name
-- i_step - step number
-- i_new_provider - new provider node
+-- Returns:
+-- 200 - success
+-- 404 - node not initialized for queue
+-- 301 - node is not root
-- ----------------------------------------------------------------------
declare
n_type text;
-- 200 - Ok
-- 304 - No such queue
-- 406 - That is a provider
+-- Calls:
+-- None
+-- Tables directly manipulated:
+-- None
------------------------------------------------------------------------
declare
_is_local boolean;
-- Get subscriber list for the local node.
--
-- It may be out-of-date, due to in-progress
--- administrative change. Node's local provider info
+-- administrative change.
+-- Node's local provider info ( pgq_node.get_node_info() or pgq_node.get_worker_state(1) )
-- is the authoritative source.
--
-- Parameters:
--
-- Parameters:
-- i_queue_name - queue name
+-- Returns:
+-- true - if this this the leaf node for queue
-- ----------------------------------------------------------------------
declare
res bool;
--
-- Parameters:
-- i_queue_name - queue name
+-- Returns:
+-- true - if this this the root node for queue
-- ----------------------------------------------------------------------
declare
res bool;
--
-- Parameters:
-- i_queue_name - queue name
+--
+-- Returns:
+-- 200 - success
+-- 404 - node not initialized for queue
+-- 301 - node is not branch
-- ----------------------------------------------------------------------
declare
n_name text;
--
-- Returns:
-- ret_code - error code
+-- 200 - ok
+-- 201 - already registered
+-- 401 - no such queue
-- ret_note - description
-- ----------------------------------------------------------------------
declare
-- i_queue_name - cascaded queue name
-- i_consumer_name - cascaded consumer name
-- i_tick_id - tick id
+-- Returns:
+-- 200 - ok
+-- 404 - consumer not known
-- ----------------------------------------------------------------------
begin
update pgq_node.local_state
-- Function: pgq_node.set_consumer_error(3)
--
-- If batch processing fails, consumer can store it's last error in db.
+-- Returns:
+-- 100 - ok
+-- 101 - consumer not known
-- ----------------------------------------------------------------------
begin
update pgq_node.local_state
-- i_queue_name - cascaded queue name
-- i_consumer_name - cascaded consumer name
-- i_paused - new flag state
+-- Returns:
+-- 200 - ok
+-- 201 - already paused
+-- 404 - consumer not found
-- ----------------------------------------------------------------------
declare
old_flag boolean;
-- i_uptodate - new flag state
--
-- Returns:
--- nothing
+-- 200 - ok
+-- 404 - consumer not known
-- ----------------------------------------------------------------------
begin
update pgq_node.local_state
-- i_watermark - partition tick_id that came inside combined-root batch
--
-- Returns:
--- nothing
+-- 200 - success
+-- 201 - no partition queue
+-- 401 - worker registration not found
-- ----------------------------------------------------------------------
declare
n record;
--
-- Notify provider about subscribers lowest watermark.
--
+-- Called on provider at interval by each worker
+--
-- Parameters:
-- i_queue_name - cascaded queue name
-- i_node_name - subscriber node name
--
-- Returns:
-- ret_code - error code
+-- 200 - ok
+-- 404 - no such queue
-- ret_note - description
-- ----------------------------------------------------------------------
begin
--
-- Return Codes:
-- 200 - Ok
--- 404 - No such set
+-- 301 - Location not found
+-- 403 - Cannot drop nodes own or parent location
-- ----------------------------------------------------------------------
declare
_queue_name text;
projects / skytools.git / commitdiff