Triggers made with plpythonu performance issue - Mailing list pgsql-general

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),
);

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;

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