Triggers made with plpythonu performance issue

[sabrina miller <sabrina.miller@xxxxxxxxx>]

Hi everybody,
My requirements was:
 + Made a table charge to be partitioned by carrier and month 
 + summarize by charges 
 + summarize by  users, 
 + each summarization must be by month and several others columns.

Here is the database

CREATE TABLE charges.charge
(
  id serial NOT NULL,
  transaction_id uuid NOT NULL,
  carrier_id integer NOT NULL,
  msisdn character varying(60) NOT NULL,
  partner_id integer NOT NULL,
  product_id integer NOT NULL,
  parent_id integer,
  retry_count integer NOT NULL,
  created_at timestamp with time zone NOT NULL,
  CONSTRAINT charge_pkey PRIMARY KEY (id)
);

CREATE TABLE charges.charge_summarized
(
  id serial NOT NULL,
  created_at timestamp with time zone NOT NULL,
  carrier_id integer NOT NULL,
  partner_id integer NOT NULL,
  product_id integer NOT NULL,
  retry_count integer NOT NULL,
  amount integer NOT NULL,
  CONSTRAINT charge_summarized_pkey PRIMARY KEY (id),
CONSTRAINT client_charge_client_id_key UNIQUE (carrier_id, partner_id, product_id, retry_count)
);

CREATE TABLE charges.client
(
  id serial NOT NULL,
  carrier_id integer NOT NULL,
  msisdn character varying(60) NOT NULL,
  collectibility numeric(5,2) NOT NULL,
  CONSTRAINT client_pkey PRIMARY KEY (id),
);


CREATE TABLE charges.client_charge
(
  id serial NOT NULL,
  client_id integer NOT NULL,
  date date NOT NULL,
  amount integer NOT NULL,
  CONSTRAINT client_charge_pkey PRIMARY KEY (id),
  CONSTRAINT client_charge_client_id_fkey FOREIGN KEY (client_id)
      REFERENCES charges.client (id) MATCH SIMPLE
      ON UPDATE NO ACTION ON DELETE NO ACTION DEFERRABLE INITIALLY DEFERRED,
  CONSTRAINT client_charge_client_id_key UNIQUE (client_id, date),
);

OK, now the functions I made to complete all this, Inserts are made only in charges.charge

create or replace function charges.insert_into_charges() returns trigger as $body1$
args = TD['new']
event = TD['event']
table_name = TD['table_name']
from datetime import datetime, timedelta
if event == 'INSERT':
  carrier = args['carrier_id']
  created_at = datetime.strptime(args['created_at'].split(" ")[0], "%Y-%m-%d") 
  month, year = created_at.month, created_at.year
  next_month, next_year = (month + 1, year) if month < 12 else (1, year+1)
  target_table_name = "charge_%s_%04d%02d" % (carrier, year, month)
  while True:
    exist_table = len(plpy.execute("select relname from pg_stat_user_tables where relname = '%s';" % target_table_name))
    if not exist_table:
      sql = """create table charges.%(target_table_name)s
        (CONSTRAINT charge_%(carrier_id)s_carrier_id_check
        CHECK (
          carrier_id = '%(carrier_id)s' AND
          created_at >= '%(from_year)s-%(from_month)s-01' AND
          created_at < '%(to_year)s-%(to_month)s-01'
          )
        )
        INHERITS (charges.charge)
        WITH (
        OIDS=FALSE
        );
       
    create trigger summarize_%(target_table_name)s
    AFTER insert
    on charges.%(target_table_name)s
    for each row
    execute procedure charges.summarize();
   
    create trigger client_charge_sum_%(target_table_name)s
    AFTER insert
    on charges.%(target_table_name)s
    for each row
    execute procedure charges.client_charge_sum();
        """ % {"carrier_id": carrier, "from_year": year, "from_month": month,
        "to_year": next_year, "to_month": next_month, "target_table_name": target_table_name}
      try:
        # multithreading could have a race condition here. Better to ask for forgiveness than permission.
        plpy.execute(sql)
      except:
        continue
    break
  keys, values = zip(*tuple([(x,y) for x,y in args.items() if y is not None]))
  sql = "insert into charges.%(target_table_name)s " \
    "(%(keys)s) VALUES (%(values)s);" % \
    {"carrier_id": carrier,
    "keys"       : ",".join(keys),
    "values"     : ",".join(["'%s'" % x for x in values]),
    "target_table_name": target_table_name
    }
 
  plpy.execute(sql)
return "SKIP"
$body1$
language plpythonu;

create or replace function charges.client_charge_sum() returns trigger as
$body3$
args = TD['new']
event = TD['event']
table_name = TD['table_name']
if event != 'INSERT':
  return
while True:
  # to populate clients if is needed
  sql = "select id from charges.client where msisdn='%s';" % args["msisdn"]
  clients = plpy.execute(sql)
  if len(clients):
    client_id = clients[0]['id']
  else:
    sql= """INSERT INTO charges.client ( carrier_id, msisdn,collectibility )
    VALUES ( %s, '%s',0) RETURNING
    CURRVAL('charges.client_id_seq') as id;""" % (args['carrier_id'], args['msisdn'])
  try:
    client_id = plpy.execute(sql)[0]['id']
  except:
    continue
  break
group_by_data ={
"date"    : args['created_at'].split(" ")[0],
"client_id"    : str(client_id),
}
filter_string = " and ".join(["%s='%s'" % (key, value) for key, value in group_by_data.items()])
while True:
  sql = "select 1 from charges.client_charge where %s;" % (filter_string,)
  if len(plpy.execute(sql)):
    sql = "update charges.client_charge set amount=amount + 1 where %s" % filter_string
    plpy.info("update")
  else:
    field_names, field_datas = zip(*tuple(group_by_data.items()))
    field_data_string = ", ".join(["'%s'" % x for x in field_datas])
    sql = "insert into charges.client_charge (amount, %s) values (1, %s)" % (", ".join(field_names), field_data_string)
    plpy.info("insert")
  try:
    plpy.execute(sql)
  except:
    continue
  break
$body3$
language plpythonu;

create or replace function charges.summarize() returns trigger as
$body2$
args = TD['new']
event = TD['event']
table_name = TD['table_name']
if event != 'INSERT':
  return

group_by_data ={
"carrier_id" : args['carrier_id'],
"charged_at"    : args['created_at'].split(" ")[0],
"partner_id"    : args['partner_id'],
"product_id"    : args['product_id'],
"retry_count"   : args['retry_count'],
}
filter_string = " and ".join(["%s='%s'" % (key, value) for key, value in group_by_data.items()])
while True:
  sql = "select 1 from charges.charge_summarized where %s;" % (filter_string,)
  if len(plpy.execute(sql)):
    sql = "update charges.charge_summarized set amount=amount + 1 where %s" % filter_string
    plpy.info("update")
  else:
    field_names, field_datas = zip(*tuple(group_by_data.items()))
    field_data_string = ", ".join(["'%s'" % x for x in field_datas])
    sql = "insert into charges.charge_summarized (amount, %s) values (1, %s)" % (", ".join(field_names), field_data_string)
    plpy.info("insert")
  try:
    plpy.execute(sql)
  except:
    continue
  break
$body2$
language plpythonu;

And finally the trigger:

CREATE TRIGGER inserta_tg
  BEFORE INSERT
  ON charges.charge
  FOR EACH ROW
  EXECUTE PROCEDURE charges.insert_into_charges();

Doesn't sound like too much? As I say, im new and I didn't found any better. But an insert takes around 135ms in the worst case (create tables and insert rows) and about 85 ms in best case (only updates). There are something better?

Thanks in advance, Sabrina