rewrite drop_triggers() in C and adapted it to multilevel partitioning

Author: zilder

init.sql

@@ -534,35 +534,8 @@ LANGUAGE plpgsql;
  */
 CREATE OR REPLACE FUNCTION @extschema@.drop_triggers(
 	parent_relid	REGCLASS)
-RETURNS VOID AS
-$$
-DECLARE
-	triggername	TEXT;
-	rec			RECORD;
-
-BEGIN
-	triggername := @extschema@.build_update_trigger_name(parent_relid);
-
-	/* 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
-				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;
+RETURNS VOID AS 'pg_pathman', 'drop_update_triggers'
+LANGUAGE C STRICT;
 
 /*
  * Drop partitions. If delete_data set to TRUE, partitions
@@ -767,7 +740,8 @@ LANGUAGE C STRICT;
  * Get parent of pg_pathman's partition.
  */
 CREATE OR REPLACE FUNCTION @extschema@.get_parent_of_partition(
-	partition_relid		REGCLASS)
+	partition_relid		REGCLASS,
+	raise_error			BOOL DEFAULT TRUE)
 RETURNS REGCLASS AS 'pg_pathman', 'get_parent_of_partition_pl'
 LANGUAGE C STRICT;
 

src/include/partition_creation.h

@@ -86,6 +86,9 @@ void create_single_update_trigger_internal(Oid partition_relid,
 
 bool has_update_trigger_internal(Oid parent);
 
+void drop_single_update_trigger_internal(Oid relid,
+										 const char *trigname);
+
 /* Partitioning callback type */
 typedef enum
 {

src/include/utils.h

@@ -30,13 +30,14 @@ bool match_expr_to_operand(Node *expr, Node *operand);
  * Misc.
  */
 Oid get_pathman_schema(void);
-List * list_reverse(List *l);
+List *list_reverse(List *l);
 
 /*
  * Useful functions for relations.
  */
 Oid get_rel_owner(Oid relid);
-char * get_rel_name_or_relid(Oid relid);
+char *get_rel_name_or_relid(Oid relid);
+char *get_qualified_rel_name(Oid relid);
 RangeVar *makeRangeVarFromRelid(Oid relid);
 
 /*
@@ -52,13 +53,13 @@ void extract_op_func_and_ret_type(char *opname,
 /*
  * Print values and cast types.
  */
-char * datum_to_cstring(Datum datum, Oid typid);
+char *datum_to_cstring(Datum datum, Oid typid);
 Datum perform_type_cast(Datum value, Oid in_type, Oid out_type, bool *success);
 Datum extract_binary_interval_from_text(Datum interval_text,
 										Oid part_atttype,
 										Oid *interval_type);
-char ** deconstruct_text_array(Datum array, int *array_size);
-RangeVar ** qualified_relnames_to_rangevars(char **relnames, size_t nrelnames);
+char **deconstruct_text_array(Datum array, int *array_size);
+RangeVar **qualified_relnames_to_rangevars(char **relnames, size_t nrelnames);
 
 
 #endif /* PATHMAN_UTILS_H */

src/partition_creation.c

@@ -1741,7 +1741,7 @@ build_partitioning_expression(Oid parent_relid,
 
 /*
  * -------------------------
- *  Update trigger creation
+ *  Update triggers management
  * -------------------------
  */
 
@@ -1774,6 +1774,8 @@ create_single_update_trigger_internal(Oid partition_relid,
 
 	(void) CreateTrigger(stmt, NULL, InvalidOid, InvalidOid,
 						 InvalidOid, InvalidOid, false);
+
+	CommandCounterIncrement();
 }
 
 /* Check if relation has pg_pathman's update trigger */
@@ -1816,3 +1818,14 @@ has_update_trigger_internal(Oid parent_relid)
 
 	return res;
 }
+
+void
+drop_single_update_trigger_internal(Oid relid,
+									const char *trigname)
+{
+	Oid				trigoid;
+
+	trigoid = get_trigger_oid(relid, trigname, true);
+	if (OidIsValid(trigoid))
+		RemoveTriggerById(trigoid);
+}

src/partition_filter.c

@@ -478,7 +478,8 @@ select_partition_for_insert(ExprContext *econtext, ExprState *expr_state,
 		Assert(rri_holder->expr_state != NULL);
 
 		/* Recursively search for subpartitions */
-		rri_holder = select_partition_for_insert(econtext, rri_holder->expr_state,
+		rri_holder = select_partition_for_insert(econtext,
+												 rri_holder->expr_state,
 												 subprel,
 												 parts_storage,
 												 estate);

src/pl_funcs.c

@@ -70,6 +70,7 @@ PG_FUNCTION_INFO_V1( invoke_on_partition_created_callback );
 PG_FUNCTION_INFO_V1( check_security_policy );
 
 PG_FUNCTION_INFO_V1( create_update_triggers );
+PG_FUNCTION_INFO_V1( drop_update_triggers );
 PG_FUNCTION_INFO_V1( pathman_update_trigger_func );
 PG_FUNCTION_INFO_V1( create_single_update_trigger );
 PG_FUNCTION_INFO_V1( has_update_trigger );
@@ -113,6 +114,9 @@ static void pathman_update_trigger_func_move_tuple(Relation source_rel,
 												   HeapTuple old_tuple,
 												   HeapTuple new_tuple);
 
+static void create_update_triggers_internal(Oid relid);
+static void drop_update_triggers_internal(Oid relid);
+
 static void collect_update_trigger_columns(Oid relid, List **columns);
 static Oid find_target_partition(Relation source_rel, HeapTuple tuple);
 static Oid find_topmost_parent(Oid partition);
@@ -150,6 +154,7 @@ get_parent_of_partition_pl(PG_FUNCTION_ARGS)
 	Oid					partition = PG_GETARG_OID(0);
 	PartParentSearch	parent_search;
 	Oid					parent;
+	bool				emit_error = PG_GETARG_BOOL(1);
 
 	/* Fetch parent & write down search status */
 	parent = get_parent_of_partition(partition, &parent_search);
@@ -160,14 +165,13 @@ get_parent_of_partition_pl(PG_FUNCTION_ARGS)
 	/* It must be parent known by pg_pathman */
 	if (parent_search == PPS_ENTRY_PART_PARENT)
 		PG_RETURN_OID(parent);
-	else
-	{
+
+	if (emit_error)
 		ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 						errmsg("\"%s\" is not a partition",
 							   get_rel_name_or_relid(partition))));
 
-		PG_RETURN_NULL();
-	}
+	PG_RETURN_NULL();
 }
 
 /*
@@ -1480,36 +1484,72 @@ pathman_update_trigger_func_move_tuple(Relation source_rel,
 	FreeTupleDesc(target_tupdesc);
 }
 
-/* Create UPDATE triggers for all partitions */
+/*
+ * Create UPDATE triggers for all partitions and subpartitions
+ */
 Datum
 create_update_triggers(PG_FUNCTION_ARGS)
 {
-	Oid						parent = PG_GETARG_OID(0);
+	Oid						relid = PG_GETARG_OID(0),
+							parent;
+	PartParentSearch		parent_search;
+
+	/*
+	 * If table has parent then we should check that parent has update trigger.
+	 * In the ideal world this error should never be thrown since we create and
+	 * drop update triggers for the whole partitions tree and not its parts
+	 */
+	parent = get_parent_of_partition(relid, &parent_search);
+	if (parent_search == PPS_ENTRY_PART_PARENT)
+		if(!has_update_trigger_internal(parent))
+			ereport(ERROR,
+					(errmsg("Parent table must have an update trigger"),
+					 errhint("Try to perform SELECT %s.create_update_triggers('%s');",
+					 		 get_namespace_name(get_pathman_schema()),
+					 		 get_qualified_rel_name(parent))));
+	
+	/* Recursively add triggers */
+	create_update_triggers_internal(relid);
+	PG_RETURN_VOID();
+}
+
+/*
+ * Create UPDATE triggers recursively
+ */
+static void
+create_update_triggers_internal(Oid relid)
+{
 	Oid					   *children;
 	const char			   *trigname;
 	const PartRelationInfo *prel;
 	uint32					i;
 	List				   *columns = NIL;
 
 	/* Check that table is partitioned */
-	prel = get_pathman_relation_info(parent);
-	shout_if_prel_is_invalid(parent, prel, PT_ANY);
+	prel = get_pathman_relation_info(relid);
+	/* TODO: check this only for topmost relid? */
+	// shout_if_prel_is_invalid(relid, prel, PT_ANY);
+	if (!prel)
+		return;
 
-	/* Acquire trigger and attribute names */
-	trigname = build_update_trigger_name_internal(parent);
+	/* Acquire trigger name */
+	trigname = build_update_trigger_name_internal(relid);
 
 	/* Create trigger for parent */
-	collect_update_trigger_columns(parent, &columns);
-	create_single_update_trigger_internal(parent, trigname, columns);
+	collect_update_trigger_columns(relid, &columns);
+	create_single_update_trigger_internal(relid, trigname, columns);
 
 	/* Fetch children array */
 	children = PrelGetChildrenArray(prel);
 
 	/* Create triggers for each partition */
 	for (i = 0; i < PrelChildrenCount(prel); i++)
+	{
 		create_single_update_trigger_internal(children[i], trigname, columns);
 
-	PG_RETURN_VOID();
+		/* Perform the same procedure on subpartitions */
+		create_update_triggers_internal(children[i]);
+	}
 }
 
 static void
@@ -1558,6 +1598,67 @@ create_single_update_trigger(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+/*
+ * Drop UPDATE triggers for all partitions and subpartitions
+ */
+Datum
+drop_update_triggers(PG_FUNCTION_ARGS)
+{
+	Oid					relid = PG_GETARG_OID(0),
+						parent;
+	PartParentSearch	parent_search;
+
+	/*
+	 * We can drop triggers only if relid is the topmost parent table (or if
+	 * its parent doesn't have update triggers (which should never happen in
+	 * the ideal world)
+	 */
+	parent = get_parent_of_partition(relid, &parent_search);
+	if (parent_search == PPS_ENTRY_PART_PARENT)
+		if(has_update_trigger_internal(parent))
+			ereport(ERROR,
+					(errmsg("Parent table must not have an update trigger"),
+					 errhint("Try to perform SELECT %s.drop_triggers('%s');",
+							 get_namespace_name(get_pathman_schema()),
+							 get_qualified_rel_name(parent))));
+
+	/* Recursively drop triggers */
+	drop_update_triggers_internal(relid);
+	PG_RETURN_VOID();
+}
+
+static void
+drop_update_triggers_internal(Oid relid)
+{
+	Oid					   *children;
+	const char			   *trigname;
+	const PartRelationInfo *prel;
+	uint32					i;
+
+	prel = get_pathman_relation_info(relid);
+	if (!prel)
+		return;
+
+	/* Acquire trigger name */
+	trigname = build_update_trigger_name_internal(relid);
+
+	/* Fetch children array */
+	children = PrelGetChildrenArray(prel);
+
+	/* Drop triggers on partitions */
+	for (i = 0; i < PrelChildrenCount(prel); i++)
+	{
+		drop_single_update_trigger_internal(children[i], trigname);
+
+		/* Recursively drop triggers on subpartitions */
+		drop_update_triggers_internal(children[i]);
+
+	}
+
+	/* Drop trigger on parent */
+	drop_single_update_trigger_internal(relid, trigname);
+}
+
 /* Check if relation has pg_pathman's update trigger */
 Datum
 has_update_trigger(PG_FUNCTION_ARGS)

src/pl_range_funcs.c

@@ -64,7 +64,6 @@ static void modify_range_constraint(Oid partition_relid,
 									Oid expression_type,
 									const Bound *lower,
 									const Bound *upper);
-static char *get_qualified_rel_name(Oid relid);
 static void drop_table_by_oid(Oid relid);
 static bool interval_is_trivial(Oid atttype,
 								Datum interval,
@@ -1156,19 +1155,6 @@ check_range_adjacence(Oid cmp_proc, Oid collid, List *ranges)
 	}
 }
 
-/*
- * Return palloced fully qualified relation name as a cstring
- */
-static char *
-get_qualified_rel_name(Oid relid)
-{
-	Oid nspid = get_rel_namespace(relid);
-
-	return psprintf("%s.%s",
-					quote_identifier(get_namespace_name(nspid)),
-					quote_identifier(get_rel_name(relid)));
-}
-
 /*
  * Drop table using it's Oid
  */

src/utils.c

@@ -193,6 +193,19 @@ get_rel_name_or_relid(Oid relid)
 	return relname;
 }
 
+/*
+ * Return palloced fully qualified relation name as a cstring
+ */
+char *
+get_qualified_rel_name(Oid relid)
+{
+	Oid nspid = get_rel_namespace(relid);
+
+	return psprintf("%s.%s",
+					quote_identifier(get_namespace_name(nspid)),
+					quote_identifier(get_rel_name(relid)));
+}
+
 RangeVar *
 makeRangeVarFromRelid(Oid relid)
 {