fix detach_range_partition()

Author: zilder

init.sql

@@ -930,7 +930,7 @@ RETURNS BOOLEAN AS 'pg_pathman', 'is_operator_supported'
 LANGUAGE C;
 
 /*
- * Referential integrity triggers
+ * Referential integrity funcs and triggers
  */
 CREATE OR REPLACE FUNCTION @extschema@.create_fk(
 	fk_table	REGCLASS,
@@ -939,6 +939,12 @@ CREATE OR REPLACE FUNCTION @extschema@.create_fk(
 RETURNS VOID AS 'pg_pathman', 'create_fk_constraint'
 LANGUAGE C STRICT;
 
+CREATE OR REPLACE FUNCTION @extschema@.prepare_partition_drop(
+	parent		REGCLASS,
+	partition	REGCLASS)
+RETURNS VOID AS 'pg_pathman', 'prepare_partition_drop'
+LANGUAGE C STRICT;
+
 CREATE OR REPLACE FUNCTION @extschema@.pathman_fkey_check_ins()
 RETURNS TRIGGER AS 'pg_pathman', 'pathman_fkey_check_ins'
 LANGUAGE C STRICT;

range.sql

@@ -489,7 +489,6 @@ CREATE OR REPLACE FUNCTION @extschema@.generate_bounds(
 RETURNS ANYARRAY AS 'pg_pathman', 'generate_bounds'
 LANGUAGE C;
 
-
 /*
  * Split RANGE partition
  */
@@ -501,98 +500,6 @@ CREATE OR REPLACE FUNCTION @extschema@.split_range_partition(
 RETURNS REGCLASS AS 'pg_pathman', 'split_range_partitions'
 LANGUAGE C;
 
-
-CREATE OR REPLACE FUNCTION @extschema@._split_range_partition(
-	partition_relid	REGCLASS,
-	split_value		ANYELEMENT,
-	partition_name	TEXT DEFAULT NULL,
-	tablespace		TEXT DEFAULT NULL,
-	OUT p_range		ANYARRAY)
-RETURNS ANYARRAY AS
-$$
-DECLARE
-	v_parent		REGCLASS;
-	v_attname		TEXT;
-	v_atttype		REGTYPE;
-	v_cond			TEXT;
-	v_new_partition	TEXT;
-	v_part_type		INTEGER;
-	v_check_name	TEXT;
-
-BEGIN
-	v_parent = @extschema@.get_parent_of_partition(partition_relid);
-
-	/* Acquire lock on parent */
-	PERFORM @extschema@.lock_partitioned_relation(v_parent);
-
-	/* Acquire data modification lock (prevent further modifications) */
-	PERFORM @extschema@.prevent_relation_modification(partition_relid);
-
-	v_atttype = @extschema@.get_partition_key_type(v_parent);
-
-	SELECT attname, parttype
-	FROM @extschema@.pathman_config
-	WHERE partrel = v_parent
-	INTO v_attname, v_part_type;
-
-	/* Check if this is a RANGE partition */
-	IF v_part_type != 2 THEN
-		RAISE EXCEPTION '"%" is not a RANGE partition', partition_relid::TEXT;
-	END IF;
-
-	/* Get partition values range */
-	EXECUTE format('SELECT @extschema@.get_part_range($1, NULL::%s)',
-				   @extschema@.get_base_type(v_atttype)::TEXT)
-	USING partition_relid
-	INTO p_range;
-
-	IF p_range IS NULL THEN
-		RAISE EXCEPTION 'could not find specified partition';
-	END IF;
-
-	/* Check if value fit into the range */
-	IF p_range[1] > split_value OR p_range[2] <= split_value
-	THEN
-		RAISE EXCEPTION 'specified value does not fit into the range [%, %)',
-			p_range[1], p_range[2];
-	END IF;
-
-	/* Create new partition */
-	v_new_partition := @extschema@.create_single_range_partition(v_parent,
-																 split_value,
-																 p_range[2],
-																 partition_name,
-																 tablespace);
-
-	/* Copy data */
-	v_cond := @extschema@.build_range_condition(v_new_partition::regclass,
-												v_attname, split_value, p_range[2]);
-	EXECUTE format('WITH part_data AS (DELETE FROM %s WHERE %s RETURNING *)
-					INSERT INTO %s SELECT * FROM part_data',
-				   partition_relid::TEXT,
-				   v_cond,
-				   v_new_partition);
-
-	/* Alter original partition */
-	v_cond := @extschema@.build_range_condition(partition_relid::regclass,
-												v_attname, p_range[1], split_value);
-	v_check_name := @extschema@.build_check_constraint_name(partition_relid, v_attname);
-
-	EXECUTE format('ALTER TABLE %s DROP CONSTRAINT %s',
-				   partition_relid::TEXT,
-				   v_check_name);
-
-	EXECUTE format('ALTER TABLE %s ADD CONSTRAINT %s CHECK (%s)',
-				   partition_relid::TEXT,
-				   v_check_name,
-				   v_cond);
-
-	/* Tell backend to reload configuration */
-	PERFORM @extschema@.on_update_partitions(v_parent);
-END
-$$
-LANGUAGE plpgsql;
-
 /*
  * Merge multiple partitions. All data will be copied to the first one.
  * The rest of partitions will be dropped.
@@ -1088,6 +995,8 @@ BEGIN
 		RAISE EXCEPTION 'table "%" is not partitioned', parent_relid::TEXT;
 	END IF;
 
+	PERFORM @extschema@.prepare_partition_drop(parent_relid, partition_relid);
+
 	/* Remove inheritance */
 	EXECUTE format('ALTER TABLE %s NO INHERIT %s',
 				   partition_relid::TEXT,

src/include/ref_integrity.h

@@ -14,7 +14,7 @@ HeapTuple get_index_for_key(Relation rel, AttrNumber attnum, Oid *index_id);
 List *get_ri_triggers_list(Oid relid, Oid constr);
 void ri_preparePartitionDrop(Oid parent,
 							 Relation partition,
-							 bool check_referencies);
+							 bool check_references);
 void ri_checkReferences(Relation partition, Oid constraintOid);
 void enable_ri_triggers(void);
 void disable_ri_triggers(void);

src/pl_funcs.c

@@ -15,6 +15,7 @@
 #include "pathman.h"
 #include "partition_creation.h"
 #include "partition_filter.h"
+#include "ref_integrity.h"
 #include "relation_info.h"
 #include "xact_handling.h"
 
@@ -84,6 +85,7 @@ PG_FUNCTION_INFO_V1( create_update_triggers );
 PG_FUNCTION_INFO_V1( update_trigger_func );
 PG_FUNCTION_INFO_V1( create_single_update_trigger );
 PG_FUNCTION_INFO_V1( is_update_trigger_enabled );
+PG_FUNCTION_INFO_V1( prepare_partition_drop );
 
 
 /*
@@ -1152,3 +1154,17 @@ is_update_trigger_enabled(PG_FUNCTION_ARGS)
 {
 	PG_RETURN_BOOL(is_update_trigger_enabled_internal(PG_GETARG_OID(0)));
 }
+
+Datum
+prepare_partition_drop(PG_FUNCTION_ARGS)
+{
+	Oid			parent = PG_GETARG_OID(0);
+	Oid			partition = PG_GETARG_OID(1);
+	Relation	partition_rel;
+
+	partition_rel = heap_open(partition, AccessExclusiveLock);
+	ri_preparePartitionDrop(parent, partition_rel, true);
+	heap_close(partition_rel, AccessExclusiveLock);
+
+	PG_RETURN_VOID();
+}

src/pl_range_funcs.c

@@ -11,10 +11,10 @@
 #include "init.h"
 #include "pathman.h"
 #include "partition_creation.h"
+#include "ref_integrity.h"
 #include "relation_info.h"
 #include "utils.h"
 #include "xact_handling.h"
-#include "ref_integrity.h"
 
 #include "access/xact.h"
 #include "catalog/namespace.h"
@@ -40,7 +40,7 @@ static ArrayType *construct_infinitable_array(Bound *elems,
 											  bool elmbyval,
 											  char elmalign);
 static void check_range_adjacence(Oid cmp_proc, Oid collid, List *ranges);
-static char *build_range_condition_internal(Oid relid, const char *attname,
+static char *build_range_condition_internal(Oid relid, char *attname,
 											const Bound *min, const Bound *max,
 											Oid bounds_type);
 static void merge_range_partitions_internal(Oid parent,
@@ -380,7 +380,7 @@ Datum
 build_range_condition(PG_FUNCTION_ARGS)
 {
 	Oid			relid = PG_GETARG_OID(0);
-	const char *attname = TextDatumGetCString(PG_GETARG_TEXT_P(1));
+	char	   *attname = TextDatumGetCString(PG_GETARG_TEXT_P(1));
 
 	Bound		min,
 				max;
@@ -395,19 +395,13 @@ build_range_condition(PG_FUNCTION_ARGS)
 				MakeBoundInf(PLUS_INFINITY) :
 				MakeBound(PG_GETARG_DATUM(3));
 
-	// con = build_range_check_constraint(relid, text_to_cstring(attname),
-	// 								   &min, &max,
-	// 								   bounds_type);
-
-	// result = deparse_constraint(relid, con->raw_expr);
 	result = build_range_condition_internal(relid, attname,
 											&min, &max, bounds_type);
-
 	PG_RETURN_TEXT_P(cstring_to_text(result));
 }
 
 static char *
-build_range_condition_internal(Oid relid, const char *attname,
+build_range_condition_internal(Oid relid, char *attname,
 							   const Bound *min, const Bound *max,
 							   Oid bounds_type)
 {
@@ -610,33 +604,30 @@ merge_range_partitions_internal(Oid parent, Oid *parts, uint32 nparts)
 Datum
 split_range_partitions(PG_FUNCTION_ARGS)
 {
-	Oid			partition = PG_GETARG_OID(0);
-	Datum		value = PG_GETARG_DATUM(1);
-	Oid			value_type = get_fn_expr_argtype(fcinfo->flinfo, 1);
-	// char	   *new_name = TextDatumGetCString(PG_GETARG_TEXT_P(2));
-	Oid			new_partition;
-	// char	   *new_partition_name = NULL;
+	Oid				partition = PG_GETARG_OID(0);
+	Datum			value = PG_GETARG_DATUM(1);
+	Oid				value_type = get_fn_expr_argtype(fcinfo->flinfo, 1);
+	Oid				new_partition;
 
-		/* Optional: name & tablespace */
-	RangeVar   *new_partition_rv = NULL;
-	char	   *new_partition_ts = NULL;
+	/* Optional: name & tablespace */
+	RangeVar	   *new_partition_rv = NULL;
+	char		   *new_partition_ts = NULL;
 
 	const PartRelationInfo *prel;
-	Oid			parent;
-	char *attname;
-	RangeEntry *rentry = NULL;
-	FmgrInfo	cmp_finfo;
-	Bound		fake_value_bound;
-	RangeEntry *ranges;
-	Bound min, max;
-
-	List *ri_constr, *ri_relids;
-	char	   *bound;
-	int i;
-	PartParentSearch		parent_search;
+	Oid				parent;
+	char		   *attname;
+	RangeEntry	   *rentry = NULL;
+	FmgrInfo		cmp_finfo;
+	Bound			fake_value_bound;
+	RangeEntry	   *ranges;
+	Bound			min, max;
+
+	List		   *ri_constr,
+				   *ri_relids;
+	char		   *bound;
+	int				i;
+	PartParentSearch parent_search;
 
-	Relation partition_rel;
-	Constraint *check;
 	AttrNumber attnum;
 	char *query;
 
@@ -648,8 +639,6 @@ split_range_partitions(PG_FUNCTION_ARGS)
 		partition_name = PG_GETARG_TEXT_P(2);
 		qualified_name = textToQualifiedNameList(partition_name);
 		new_partition_rv = makeRangeVarFromNameList(qualified_name);
-
-		// new_partition_rv = TextDatumGetCString(PG_GETARG_TEXT_P(2));
 	}
 
 	if (!PG_ARGISNULL(3))
@@ -700,7 +689,6 @@ split_range_partitions(PG_FUNCTION_ARGS)
 	pathman_get_fkeys(parent, &ri_constr, &ri_relids);
 
 	/* Create new partition */
-	/* TODO: partition name and tablespace instead of NULL */
 	new_partition = create_single_range_partition_internal(parent,
 												  &fake_value_bound,
 												  &max,
@@ -714,10 +702,10 @@ split_range_partitions(PG_FUNCTION_ARGS)
 
 	/* Copy data */
 	bound = build_range_condition_internal(partition,
-								   attname,
-								   &fake_value_bound,
-								   &max,
-								   prel->atttype);
+										   attname,
+										   &fake_value_bound,
+										   &max,
+										   prel->atttype);
 
 	if (SPI_connect() != SPI_OK_CONNECT)
 		elog(ERROR, "could not connect using SPI");
@@ -741,21 +729,6 @@ split_range_partitions(PG_FUNCTION_ARGS)
 							attname, attnum, prel->atttype,
 							&min, &fake_value_bound);
 
-	// drop_check_constraint(partition, attnum);
-	// check = build_range_check_constraint(partition,
-	// 							 attname,
-	// 							 &min,
-	// 							 &fake_value_bound,
-	// 							 prel->atttype);
-
-	// /* TODO: think about locking */
-	// partition_rel = heap_open(partition, ExclusiveLock);
-
-	// AddRelationNewConstraints(partition_rel, NIL,
-	// 						  list_make1(check),
-	// 						  false, true, true);
-	// heap_close(partition_rel, AccessExclusiveLock);
-
 	/* get_pathman_relation_info() will refresh this entry */
 	invalidate_pathman_relation_info(parent, NULL);
 
@@ -824,7 +797,7 @@ drop_range_partition_expand_next(PG_FUNCTION_ARGS)
 								&next->max);
 	}
 
-	/* Perform checks and preparations regarding to reference integrity */
+	/* Perform preparations regarding reference integrity */
 	ri_preparePartitionDrop(parent, rel, false);
 	CommandCounterIncrement();
 

src/ref_integrity.c

@@ -1900,7 +1900,7 @@ get_ri_triggers_list(Oid relid, Oid constr)
 void
 ri_preparePartitionDrop(Oid parent,
 						Relation partition,
-						bool check_referencies)
+						bool check_references)
 {
 	List	   *ri_constr;
 	List	   *ri_refrelids;
@@ -1924,7 +1924,7 @@ ri_preparePartitionDrop(Oid parent,
 		shout_if_prel_is_invalid(parent, prel, PT_INDIFFERENT);
 
 		/* Check if there are references in FK table */
-		if (check_referencies)
+		if (check_references)
 			ri_checkReferences(partition, constr);
 
 		/*