CREATE TABLE obj_base
(
 obj_id bigint NOT NULL,
 obj_created timestamp with time zone NOT NULL DEFAULT now(),
 obj_modified timestamp with time zone NOT NULL DEFAULT now(),
 obj_status_did smallint NOT NULL DEFAULT 1,

 CONSTRAINT pk_obj_base PRIMARY KEY (obj_id)
);


COMMENT ON TABLE obj_base IS 'Abstract base obj';


CREATE OR REPLACE FUNCTION t_obj_modified()
 RETURNS trigger AS
$BODY$
-- Trigger function to set modification field
BEGIN
   SELECT INTO NEW.obj_modified CURRENT_TIMESTAMP;
   RETURN NEW;
END;
$BODY$
 LANGUAGE 'plpgsql' VOLATILE
 COST 1;


COMMENT ON FUNCTION t_obj_modified() IS 'Trigger function to set
modification field';


CREATE SCHEMA activity;


CREATE TABLE activity.stream
(
 obj_id bigserial NOT NULL,

 as_generator_obj_id bigint NOT NULL,
 as_type_did bigint NOT NULL,
 as_obj_ids bigint[],
 as_data hstore,

 CONSTRAINT pk_activity_stream PRIMARY KEY (obj_id)
) INHERITS (obj_base);


COMMENT ON TABLE activity.stream IS 'Activity streams';


CREATE TRIGGER t_activity_stream__modified_bu
 BEFORE UPDATE
 ON activity.stream
 FOR EACH ROW
 EXECUTE PROCEDURE t_obj_modified();


CREATE INDEX i_activity_stream__generator_created ON activity.stream
USING btree (as_generator_obj_id, obj_created DESC);


CREATE OR REPLACE FUNCTION activity.t_stream__partition_redirect()
 RETURNS trigger AS
$BODY$
""" Trigger function to redirect inserts into appropriate partitions
for activity.stream table.

activity.stream tables partitioning is based on obj_created column.
New partition is created
when first row insert appears after midnight in UTC. All partition
constraints are also in UTC
so it is guaranty that every row will be inserted into correct
partition despite daylight time
changes. Partition name template is:

   activity.stream__pYYYYMMDD

For each partition it creates an index:

   i_activity_stream__pYYYYMMDD__generator_created

The functions body, query plans and other stuff are stored into
session data (SD) to get
better performance.

TODO: Should think about generic partitioning solution.
"""

# Declare session data namespace.
sdNamespace = 'activity.t_stream__partition_redirect'

# If the function is not in session data do all the preparations and
put it there.
if sdNamespace not in SD:

   import re
   from datetime import datetime, timedelta, tzinfo

   def pgTimestampToPyDatetime(timestamp):
       """ Parse ISO 8601 format timestamp to datetime.

       YYYY-MM-DD[ T]HH:MM[:SS[.ss]][-+HH[:MM]]

       Based on solution of Marko Kreen.
       """

       class FixedOffsetTimezone(tzinfo):
           """Fixed offset in minutes east from UTC."""

           def __init__(self, offset, name):
               self.__offset = timedelta(minutes = offset)
               self.__name = name

           def utcoffset(self, dt):
               return self.__offset

           def tzname(self, dt):
               return self.__name

           def dst(self, dt):
               return timedelta(0)

       # Prerare ISO8601 regexp to parse PG timestamp.
       regexpKey = '%s:iso8601Regexp' % sdNamespace
       if regexpKey not in SD:
           iso8601Regexp = SD[regexpKey] = re.compile(r"""
               \s*
               (?P<year> \d\d\d\d) [-] (?P<month> \d\d) [-] (?P<day> \d\d) [ T]
               (?P<hour> \d\d) [:] (?P<min> \d\d)
               (?: [:] (?P<sec> \d\d ) (?: [.,] (?P<ss> \d+))? )?
               (?: \s* (?P<tzsign> [-+]) (?P<tzhr> \d\d) (?: [:]?(?P<tzmin> \d\d))? )?
               \s* $
               """, re.X)
       else:
           iso8601Regexp = SD[regexpKey]

       # Try to parse, raise error on failure.
       matches = iso8601Regexp.match(timestamp)
       if not matches:
           plpy.error('Timestamp not in ISO8601 format: %s' % repr(timestamp))

       # Process timezone if it is declared.
       timezone = None
       if matches.group('tzsign'):
           tzOffsetMinutes = int(matches.group('tzhr')) * 60

           if matches.group('tzmin'):
               tzOffsetMinutes += int(matches.group('tzmin'))

           if matches.group('tzsign') == '-':
               tzOffsetMinutes = -tzOffset

           tzName = '%s%s:%s' % (
               matches.group('tzsign'), matches.group('tzhr'),
               matches.group('tzmin') or '00')

           timezone = FixedOffsetTimezone(tzOffsetMinutes, tzName)

       # Create datetime object and return it.
       return datetime(
           int(matches.group('year')),
           int(matches.group('month')),
           int(matches.group('day')),
           int(matches.group('hour')),
           int(matches.group('min')),
           matches.group('sec') and int(matches.group('sec')) or 0,
           matches.group('ss') and int(matches.group('ss').ljust(6, '0')) or 0,
           timezone)

   def requireCorrectExecutionContext():
       """ Check if it is BEFORE INSERT trigger, raise error if not. """

       if TD['event'] != 'INSERT' or TD['when'] != 'BEFORE':
           plpy.error('This procedure is designed to be called *only* from a BEFORE INSERT trigger.')

   def createPartitionIfDoesNotExists((partitionName, pgSameDay, pgNextDay)):
       """ Check if nesessary partition has already been created, create if not. """

       # Check partition existence and return if true.
       planKey = '%s:partitionExistsPlan_%s' % (sdNamespace, partitionName)
       if planKey not in SD:
           partitionExistsPlan = SD[planKey] = plpy.prepare(
               "SELECT 1 FROM pg_stat_user_tables WHERE schemaname = 'activity' AND relname = $1",
               ('text',))
       else:
           partitionExistsPlan = SD[planKey]
       if len(plpy.execute(partitionExistsPlan, ('stream__p%s' % partitionName,))):
           return

       # Create partition...
       plpy.execute("""
           CREATE TABLE activity.stream__p%(partition)s (
               CONSTRAINT pk_activity_stream__p%(partition)s PRIMARY KEY (obj_id),
               CHECK (obj_created >= '%(same)s' AND obj_created < '%(next)s')
           ) INHERITS (activity.stream);
           """ % {'partition': partitionName, 'same': pgSameDay, 'next': pgNextDay})

       # ...and its index.
       plpy.execute("""
           CREATE INDEX i_activity_stream__p%(partition)s__generator_created
           ON activity.stream__p%(partition)s
           USING btree (as_generator_obj_id, obj_created DESC);
           """ % {'partition': partitionName})

   def insertIntoPartition((partitionName, pgSameDay, pgNextDay), newRow):
       """ Do insert into specified partition. """

       planKey = '%s:insertIntoPartitionPlan:%s' % (sdNamespace, partitionName)
       if planKey not in SD:
           insertIntoPartitionPlan = SD[planKey] = plpy.prepare("""
               INSERT INTO activity.stream__p%s (
                   obj_id, obj_created, obj_modified, obj_status_did,
                   as_generator_obj_id, as_type_did, as_obj_ids, as_data
               ) VALUES (
                   $1, $2::timestamptz, $3::timestamptz, $4,
                   $5, $6, $7::bigint[], $8::hstore);
               """ % partitionName, (
                   'int8', 'text', 'text', 'int2',
                   'int8', 'int8', 'text', 'text'))
       else:
           insertIntoPartitionPlan = SD[planKey]

       plpy.execute(
           insertIntoPartitionPlan,
           [newRow[k] for k in ('obj_id', 'obj_created', 'obj_modified', 'obj_status_did',
                                'as_generator_obj_id', 'as_type_did', 'as_obj_ids', 'as_data')])

   def getPartitionMetadata(row):
       """ Define partition identifier and borders using obj_created column. """

       pyDatetime = pgTimestampToPyDatetime(row['obj_created'])

       utcSameYmdTuple = pyDatetime.utctimetuple()[:3]
       utcNextYmdTuple = (pyDatetime + timedelta(1)).utctimetuple()[:3]

       return (
           '%04d%02d%02d' % utcSameYmdTuple,
           '%04d-%02d-%02d UTC' % utcSameYmdTuple,
           '%04d-%02d-%02d UTC' % utcNextYmdTuple)

   def main(newRow):
       """ The main function. It is assumed to be cached in session data. """

       requireCorrectExecutionContext()

       partitionMetadata = getPartitionMetadata(newRow)

       createPartitionIfDoesNotExists(partitionMetadata)
       insertIntoPartition(partitionMetadata, newRow)

   # Cache body in SD
   SD[sdNamespace] = main

try:
   SD[sdNamespace](TD['new'])
except Exception, e:
   import traceback
   plpy.info(traceback.format_exc())

return 'SKIP'

$BODY$
 LANGUAGE 'plpythonu' VOLATILE
 COST 1;


CREATE TRIGGER t_activity_stream__partition_redirect__bi
 BEFORE INSERT
 ON activity.stream
 FOR EACH ROW
 EXECUTE PROCEDURE activity.t_stream__partition_redirect();


DROP TABLE activity.stream__p20091125;

INSERT INTO activity.stream (obj_created, as_generator_obj_id, as_type_did)
VALUES ('2009-11-25 12:07:25.123456+03', 1234567890, 9876543210);

SELECT * FROM activity.stream__p20091125;