auto trigger creation for partitions

zilder · zilder · commit ecec00603dbb · 2017-02-21T17:57:07.000+03:00
diff --git a/expected/pathman_basic.out b/expected/pathman_basic.out
@@ -1321,6 +1321,7 @@ EXPLAIN (COSTS OFF) SELECT * FROM test.range_rel WHERE dt BETWEEN '2014-11-15' A
 (6 rows)
 
 SELECT pathman.detach_range_partition('test.range_rel_archive');
+NOTICE:  trigger "range_rel_upd_trig" for relation "test.range_rel_archive" does not exist, skipping
  detach_range_partition 
 ------------------------
  test.range_rel_archive
@@ -1734,6 +1735,7 @@ SELECT * FROM test."TeSt";
  1 | 1
 (3 rows)
 
+DROP TABLE test."TeSt" CASCADE;
 CREATE TABLE test."RangeRel" (
 	id	SERIAL PRIMARY KEY,
 	dt	TIMESTAMP NOT NULL,
@@ -2189,6 +2191,6 @@ ORDER BY partition;
 DROP TABLE test.provided_part_names CASCADE;
 NOTICE:  drop cascades to 2 other objects
 DROP SCHEMA test CASCADE;
-NOTICE:  drop cascades to 49 other objects
+NOTICE:  drop cascades to 48 other objects
 DROP EXTENSION pg_pathman CASCADE;
 DROP SCHEMA pathman CASCADE;
diff --git a/init.sql b/init.sql
@@ -562,7 +562,17 @@ LANGUAGE C;
  */
 CREATE OR REPLACE FUNCTION @extschema@.create_update_triggers(parent_relid REGCLASS)
 RETURNS VOID AS 'pg_pathman', 'create_update_triggers'
-LANGUAGE C;
+LANGUAGE C STRICT;
+
+CREATE OR REPLACE FUNCTION @extschema@.create_single_update_trigger(
+	parent_relid	REGCLASS,
+	partition_relid	REGCLASS)
+RETURNS VOID AS 'pg_pathman', 'create_single_update_trigger'
+LANGUAGE C STRICT;
+
+CREATE OR REPLACE FUNCTION @extschema@.is_update_trigger_enabled(parent_relid REGCLASS)
+RETURNS BOOL AS 'pg_pathman', 'is_update_trigger_enabled'
+LANGUAGE C STRICT;
 
 /*
  * Drop triggers
@@ -580,13 +590,22 @@ BEGIN
 
 	/* Drop trigger for each partition if exists */
 	FOR rec IN (SELECT pg_catalog.pg_inherits.* FROM pg_catalog.pg_inherits
-				JOIN pg_catalog.pg_trigger on inhrelid = tgrelid
+				JOIN pg_catalog.pg_trigger ON inhrelid = tgrelid
 				WHERE inhparent = parent_relid AND tgname = triggername)
 	LOOP
 		EXECUTE format('DROP TRIGGER IF EXISTS %s ON %s',
 					   triggername,
 					   rec.inhrelid::REGCLASS::TEXT);
 	END LOOP;
+
+	/* Drop trigger on parent */
+	IF EXISTS (SELECT * FROM pg_catalog.pg_trigger
+			   WHERE tgname = triggername AND tgrelid = parent_relid)
+	THEN
+		EXECUTE format('DROP TRIGGER IF EXISTS %s ON %s',
+					   triggername,
+					   parent_relid::TEXT);
+	END IF;
 END
 $$ LANGUAGE plpgsql STRICT;
 
diff --git a/range.sql b/range.sql
@@ -935,6 +935,12 @@ BEGIN
 	ON params.partrel = parent_relid
 	INTO v_init_callback;
 
+	/* If update trigger is enabled then create one for this partition */
+	if @extschema@.is_update_trigger_enabled(parent_relid) THEN
+		PERFORM @extschema@.create_single_update_trigger(parent_relid, partition_relid);
+	END IF;
+
+	/* Invoke an initialization callback */
 	PERFORM @extschema@.invoke_on_partition_created_callback(parent_relid,
 															 partition_relid,
 															 v_init_callback,
@@ -984,6 +990,11 @@ BEGIN
 				   partition_relid::TEXT,
 				   @extschema@.build_check_constraint_name(partition_relid, v_attname));
 
+	/* Remove update trigger */
+	EXECUTE format('DROP TRIGGER IF EXISTS %s ON %s',
+				   @extschema@.build_update_trigger_name(parent_relid),
+				   partition_relid::TEXT);
+
 	/* Invalidate cache */
 	PERFORM @extschema@.on_update_partitions(parent_relid);
 
diff --git a/sql/pathman_basic.sql b/sql/pathman_basic.sql
@@ -471,6 +471,7 @@ SELECT * FROM test."TeSt" WHERE a = 1;
 EXPLAIN (COSTS OFF) SELECT * FROM test."TeSt" WHERE a = 1;
 SELECT pathman.drop_partitions('test."TeSt"');
 SELECT * FROM test."TeSt";
+DROP TABLE test."TeSt" CASCADE;
 
 CREATE TABLE test."RangeRel" (
 	id	SERIAL PRIMARY KEY,
diff --git a/src/partition_creation.c b/src/partition_creation.c
@@ -22,12 +22,14 @@
 #include "catalog/heap.h"
 #include "catalog/pg_authid.h"
 #include "catalog/pg_proc.h"
+#include "catalog/pg_trigger.h"
 #include "catalog/pg_type.h"
 #include "catalog/toasting.h"
 #include "commands/event_trigger.h"
 #include "commands/sequence.h"
 #include "commands/tablecmds.h"
 #include "commands/tablespace.h"
+#include "commands/trigger.h"
 #include "miscadmin.h"
 #include "parser/parse_func.h"
 #include "parser/parse_relation.h"
@@ -52,9 +54,11 @@ static Oid spawn_partitions_val(Oid parent_relid,
 								Datum value,
 								Oid value_type);
 
-static void create_single_partition_common(Oid partition_relid,
+static void create_single_partition_common(Oid parent_relid,
+										   Oid partition_relid,
 										   Constraint *check_constraint,
-										   init_callback_params *callback_params);
+										   init_callback_params *callback_params,
+										   const char *attname);
 
 static Oid create_single_partition_internal(Oid parent_relid,
 											RangeVar *partition_rv,
@@ -77,6 +81,8 @@ static Constraint *make_constraint_common(char *name, Node *raw_expr);
 static Value make_string_value_struct(char *str);
 static Value make_int_value_struct(int int_val);
 
+static bool update_trigger_exists(Oid relid, char *trigname);
+
 
 /*
  * ---------------------------------------
@@ -130,9 +136,11 @@ create_single_range_partition_internal(Oid parent_relid,
 								*start_value, *end_value, value_type);
 
 	/* Add constraint & execute init_callback */
-	create_single_partition_common(partition_relid,
+	create_single_partition_common(parent_relid,
+								   partition_relid,
 								   check_constr,
-								   &callback_params);
+								   &callback_params,
+								   partitioned_column);
 
 	/* Return the Oid */
 	return partition_relid;
@@ -183,21 +191,25 @@ create_single_hash_partition_internal(Oid parent_relid,
 							   parent_relid, partition_relid);
 
 	/* Add constraint & execute init_callback */
-	create_single_partition_common(partition_relid,
+	create_single_partition_common(parent_relid,
+								   partition_relid,
 								   check_constr,
-								   &callback_params);
+								   &callback_params,
+								   partitioned_column);
 
 	/* Return the Oid */
 	return partition_relid;
 }
 
 /* Add constraint & execute init_callback */
 void
-create_single_partition_common(Oid partition_relid,
+create_single_partition_common(Oid parent_relid,
+							   Oid partition_relid,
 							   Constraint *check_constraint,
-							   init_callback_params *callback_params)
+							   init_callback_params *callback_params,
+							   const char *attname)
 {
-	Relation child_relation;
+	Relation	child_relation;
 
 	/* Open the relation and add new check constraint & fkeys */
 	child_relation = heap_open(partition_relid, AccessExclusiveLock);
@@ -209,6 +221,20 @@ create_single_partition_common(Oid partition_relid,
 	/* Make constraint visible */
 	CommandCounterIncrement();
 
+	/* Create trigger */
+	if (is_update_trigger_enabled_internal(parent_relid))
+	{
+		char	   *trigname;
+
+		trigname = build_update_trigger_name_internal(parent_relid);
+		create_single_update_trigger_internal(partition_relid,
+											  trigname,
+											  attname);
+	}
+	
+	/* Make trigger visible */
+	CommandCounterIncrement();
+
 	/* Finally invoke 'init_callback' */
 	invoke_part_callback(callback_params);
 
@@ -1649,3 +1675,82 @@ text_to_regprocedure(text *proc_signature)
 
 	return DatumGetObjectId(result);
 }
+
+/*
+ * Create trigger for partition
+ */
+void
+create_single_update_trigger_internal(Oid relid,
+									  const char *trigname,
+									  const char *attname)
+{
+	CreateTrigStmt	   *stmt;
+	List			   *func;
+
+	func = list_make2(makeString(get_namespace_name(get_pathman_schema())),
+					  makeString("update_trigger_func"));
+
+	stmt = makeNode(CreateTrigStmt);
+	stmt->trigname = (char *) trigname;
+	stmt->relation = makeRangeVarFromRelid(relid);
+	stmt->funcname = func;
+	stmt->args = NIL;
+	stmt->row = true;
+	stmt->timing = TRIGGER_TYPE_BEFORE;
+	stmt->events = TRIGGER_TYPE_UPDATE;
+	stmt->columns = list_make1(makeString((char *) attname));
+	stmt->whenClause = NULL;
+	stmt->isconstraint = false;
+	stmt->deferrable = false;
+	stmt->initdeferred = false;
+	stmt->constrrel = NULL;
+
+	(void) CreateTrigger(stmt, NULL, InvalidOid, InvalidOid,
+						 InvalidOid, InvalidOid, false);
+}
+
+/*
+ * Check if update trigger is enabled. Basicly it returns true if update
+ * trigger exists for parent table
+ */
+bool
+is_update_trigger_enabled_internal(Oid parent)
+{
+	char	   *trigname;
+
+	trigname = build_update_trigger_name_internal(parent);
+	return update_trigger_exists(parent, trigname);
+}
+
+static bool
+update_trigger_exists(Oid relid, char *trigname)
+{
+	bool		res = false;
+	Relation	tgrel;
+	SysScanDesc tgscan;
+	ScanKeyData key;
+	HeapTuple	tuple;
+
+	tgrel = heap_open(TriggerRelationId, RowExclusiveLock);
+
+	ScanKeyInit(&key,
+				Anum_pg_trigger_tgrelid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(relid));
+	tgscan = systable_beginscan(tgrel, TriggerRelidNameIndexId, true,
+								NULL, 1, &key);
+	while (HeapTupleIsValid(tuple = systable_getnext(tgscan)))
+	{
+		Form_pg_trigger pg_trigger = (Form_pg_trigger) GETSTRUCT(tuple);
+
+		if (namestrcmp(&(pg_trigger->tgname), trigname) == 0)
+		{
+			res = true;
+			break;
+		}
+	}
+	systable_endscan(tgscan);
+	heap_close(tgrel, RowExclusiveLock);
+
+	return res;
+}
diff --git a/src/partition_creation.h b/src/partition_creation.h
@@ -77,6 +77,12 @@ Node * build_raw_hash_check_tree(char *attname,
 
 void drop_check_constraint(Oid relid, AttrNumber attnum);
 
+/* Update triggers */
+void create_single_update_trigger_internal(Oid relid,
+										   const char *trigname,
+										   const char *attname);
+bool is_update_trigger_enabled_internal(Oid parent);
+
 
 /* Partitioning callback type */
 typedef enum
diff --git a/src/pl_funcs.c b/src/pl_funcs.c
