introduce new subsystem called 'xact_handling' & function get_pathman_relation_info_after_lock(), extract extract_binary_interval_from_text() from create_partitions_internal(), improve BGW startup

Author: funbringer

Makefile

@@ -3,7 +3,7 @@
 MODULE_big = pg_pathman
 OBJS = src/init.o src/relation_info.o src/utils.o src/partition_filter.o src/runtimeappend.o \
 	src/runtime_merge_append.o src/pg_pathman.o src/dsm_array.o src/rangeset.o src/pl_funcs.o \
-	src/worker.o src/hooks.o src/nodes_common.o $(WIN32RES)
+	src/worker.o src/hooks.o src/nodes_common.o src/xact_handling.o $(WIN32RES)
 
 EXTENSION = pg_pathman
 EXTVERSION = 0.1

init.sql

@@ -317,13 +317,16 @@ SET pg_pathman.enable_partitionfilter = off;
 
 
 
+/*
+ * Create DDL trigger to call pathman_ddl_trigger_func().
+ */
 CREATE EVENT TRIGGER pathman_ddl_trigger
 ON sql_drop
 EXECUTE PROCEDURE @extschema@.pathman_ddl_trigger_func();
 
 
 /*
- * Attach partitioned table
+ * Attach a previously partitioned table
  */
 CREATE OR REPLACE FUNCTION @extschema@.add_to_pathman_config(
 	parent_relid	REGCLASS,
@@ -356,6 +359,14 @@ CREATE OR REPLACE FUNCTION @extschema@.get_parent_of_partition(REGCLASS)
 RETURNS REGCLASS AS 'pg_pathman', 'get_parent_of_partition_pl'
 LANGUAGE C STRICT;
 
+/*
+ * Checks if attribute is nullable
+ */
+CREATE OR REPLACE FUNCTION @extschema@.is_attribute_nullable(
+	REGCLASS, TEXT)
+RETURNS BOOLEAN AS 'pg_pathman', 'is_attribute_nullable'
+LANGUAGE C STRICT;
+
 /*
  * Check if regclass is date or timestamp
  */
@@ -365,19 +376,18 @@ RETURNS BOOLEAN AS 'pg_pathman', 'is_date_type'
 LANGUAGE C STRICT;
 
 /*
- * Checks if attribute is nullable
+ * Returns attribute type name for relation
  */
-CREATE OR REPLACE FUNCTION @extschema@.is_attribute_nullable(
+CREATE OR REPLACE FUNCTION @extschema@.get_attribute_type_name(
 	REGCLASS, TEXT)
-RETURNS BOOLEAN AS 'pg_pathman', 'is_attribute_nullable'
+RETURNS TEXT AS 'pg_pathman', 'get_attribute_type_name'
 LANGUAGE C STRICT;
 
 /*
- * Returns attribute type name for relation
+ * Get parent of pg_pathman's partition.
  */
-CREATE OR REPLACE FUNCTION @extschema@.get_attribute_type_name(
-	REGCLASS, TEXT)
-RETURNS TEXT AS 'pg_pathman', 'get_attribute_type_name'
+CREATE OR REPLACE FUNCTION @extschema@.get_parent_of_partition(REGCLASS)
+RETURNS REGCLASS AS 'pg_pathman', 'get_parent_of_partition_pl'
 LANGUAGE C STRICT;
 
 /*
@@ -406,6 +416,16 @@ CREATE OR REPLACE FUNCTION @extschema@.build_update_trigger_func_name(
 RETURNS TEXT AS 'pg_pathman', 'build_update_trigger_func_name'
 LANGUAGE C STRICT;
 
+
+/*
+ * Lock partitioned relation to restrict concurrent modification of partitioning scheme.
+ */
+ CREATE OR REPLACE FUNCTION @extschema@.lock_partitioned_relation(
+	 REGCLASS)
+ RETURNS VOID AS 'pg_pathman', 'lock_partitioned_relation'
+ LANGUAGE C STRICT;
+
+
 /*
  * DEBUG: Place this inside some plpgsql fuction and set breakpoint.
  */

src/init.c

@@ -285,10 +285,8 @@ init_shmem_config(void)
 		 */
 		if (!IsUnderPostmaster)
 		{
-			/* Initialize locks */
-			pmstate->load_config_lock		= LWLockAssign();
-			pmstate->dsm_init_lock			= LWLockAssign();
-			pmstate->edit_partitions_lock	= LWLockAssign();
+			/* NOTE: dsm_array is redundant, hence the commented code */
+			/* pmstate->dsm_init_lock = LWLockAssign(); */
 		}
 	}
 }

src/pathman.h

@@ -69,9 +69,7 @@ Oid get_pathman_config_relid(void);
  */
 typedef struct PathmanState
 {
-	LWLock		   *dsm_init_lock,
-				   *load_config_lock,
-				   *edit_partitions_lock;
+	LWLock		   *dsm_init_lock;	/* unused */
 } PathmanState;
 
 

src/pg_pathman.c

@@ -16,6 +16,7 @@
 #include "partition_filter.h"
 #include "runtimeappend.h"
 #include "runtime_merge_append.h"
+#include "xact_handling.h"
 
 #include "postgres.h"
 #include "access/heapam.h"
@@ -60,6 +61,9 @@ static Node *wrapper_make_expression(WrapperNode *wrap, int index, bool *alwaysT
 static bool disable_inheritance_subselect_walker(Node *node, void *context);
 
 /* "Partition creation"-related functions */
+static Datum extract_binary_interval_from_text(Datum interval_text,
+											   Oid part_atttype,
+											   Oid *interval_type);
 static bool spawn_partitions(Oid partitioned_rel,
 							 Datum value,
 							 Datum leading_bound,
@@ -128,14 +132,15 @@ _PG_init(void)
 
 	if (!process_shared_preload_libraries_in_progress)
 	{
-		elog(ERROR, "Pathman module must be initialized in postmaster. "
+		elog(ERROR, "pg_pathman module must be initialized by Postmaster. "
 					"Put the following line to configuration file: "
 					"shared_preload_libraries='pg_pathman'");
 	}
 
 	/* Request additional shared resources */
 	RequestAddinShmemSpace(estimate_pathman_shmem_size());
-	RequestAddinLWLocks(3);
+
+	/* NOTE: we don't need LWLocks now. RequestAddinLWLocks(1); */
 
 	/* Assign pg_pathman's initial state */
 	temp_init_state.initialization_needed = true;
@@ -791,6 +796,57 @@ spawn_partitions(Oid partitioned_rel,		/* parent's Oid */
 	return spawned;
 }
 
+/*
+ * Convert interval from TEXT to binary form using partitioned column's type.
+ */
+static Datum
+extract_binary_interval_from_text(Datum interval_text,	/* interval as TEXT */
+								  Oid part_atttype,		/* partitioned column's type */
+								  Oid *interval_type)	/* returned value */
+{
+	Datum		interval_binary;
+	const char *interval_cstring;
+
+	interval_cstring = TextDatumGetCString(interval_text);
+
+	/* If 'part_atttype' is a *date type*, cast 'range_interval' to INTERVAL */
+	if (is_date_type_internal(part_atttype))
+	{
+		int32	interval_typmod = PATHMAN_CONFIG_interval_typmod;
+
+		/* Convert interval from CSTRING to internal form */
+		interval_binary = DirectFunctionCall3(interval_in,
+											  CStringGetDatum(interval_cstring),
+											  ObjectIdGetDatum(InvalidOid),
+											  Int32GetDatum(interval_typmod));
+		if (interval_type)
+			*interval_type = INTERVALOID;
+	}
+	/* Otherwise cast it to the partitioned column's type */
+	else
+	{
+		HeapTuple	htup;
+		Oid			typein_proc = InvalidOid;
+
+		htup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(part_atttype));
+		if (HeapTupleIsValid(htup))
+		{
+			typein_proc = ((Form_pg_type) GETSTRUCT(htup))->typinput;
+			ReleaseSysCache(htup);
+		}
+		else
+			elog(ERROR, "Cannot find input function for type %u", part_atttype);
+
+		/* Convert interval from CSTRING to 'prel->atttype' */
+		interval_binary = OidFunctionCall1(typein_proc,
+										   CStringGetDatum(interval_cstring));
+		if (interval_type)
+			*interval_type = part_atttype;
+	}
+
+	return interval_binary;
+}
+
 /*
  * Append partitions (if needed) and return Oid of the partition to contain value.
  *
@@ -808,9 +864,6 @@ create_partitions_internal(Oid relid, Datum value, Oid value_type)
 		Datum					values[Natts_pathman_config];
 		bool					isnull[Natts_pathman_config];
 
-		prel = get_pathman_relation_info(relid);
-		shout_if_prel_is_invalid(relid, prel, PT_RANGE);
-
 		/* Get both PartRelationInfo & PATHMAN_CONFIG contents for this relation */
 		if (pathman_config_contains_relation(relid, values, isnull, NULL))
 		{
@@ -820,54 +873,27 @@ create_partitions_internal(Oid relid, Datum value, Oid value_type)
 			Oid			interval_type = InvalidOid;
 			Datum		interval_binary, /* assigned 'width' of a single partition */
 						interval_text;
-			const char *interval_cstring;
 
 			FmgrInfo	interval_type_cmp;
 
-			/* Fill the FmgrInfo struct with a cmp(value, part_attribute) function */
-			fill_type_cmp_fmgr_info(&interval_type_cmp, value_type, prel->atttype);
-
-			/* Convert interval from TEXT to CSTRING */
-			interval_text = values[Anum_pathman_config_range_interval - 1];
-			interval_cstring = TextDatumGetCString(interval_text);
+			/* Fetch PartRelationInfo by 'relid' */
+			prel = get_pathman_relation_info(relid);
+			shout_if_prel_is_invalid(relid, prel, PT_RANGE);
 
 			/* Read max & min range values from PartRelationInfo */
 			min_rvalue = prel->ranges[0].min;
 			max_rvalue = prel->ranges[PrelLastChild(prel)].max;
 
-			/* If this is a *date type*, cast 'range_interval' to INTERVAL */
-			if (is_date_type_internal(value_type))
-			{
-				int32	interval_typmod = PATHMAN_CONFIG_interval_typmod;
-
-				/* Convert interval from CSTRING to internal form */
-				interval_binary = DirectFunctionCall3(interval_in,
-													  CStringGetDatum(interval_cstring),
-													  ObjectIdGetDatum(InvalidOid),
-													  Int32GetDatum(interval_typmod));
-				interval_type = INTERVALOID;
-			}
-			/* Otherwise cast it to the partitioned column's type */
-			else
-			{
-				HeapTuple	htup;
-				Oid			typein_proc = InvalidOid;
+			/* Retrieve interval as TEXT from tuple */
+			interval_text = values[Anum_pathman_config_range_interval - 1];
 
-				htup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(prel->atttype));
-				if (HeapTupleIsValid(htup))
-				{
-					typein_proc = ((Form_pg_type) GETSTRUCT(htup))->typinput;
-					ReleaseSysCache(htup);
-				}
-				else
-					elog(ERROR, "Cannot find input function for type %u",
-						 prel->atttype);
-
-				/* Convert interval from CSTRING to 'prel->atttype' */
-				interval_binary = OidFunctionCall1(typein_proc,
-												   CStringGetDatum(interval_cstring));
-				interval_type = prel->atttype;
-			}
+			/* Convert interval to binary representation */
+			interval_binary = extract_binary_interval_from_text(interval_text,
+																prel->atttype,
+																&interval_type);
+
+			/* Fill the FmgrInfo struct with a cmp(value, part_attribute) function */
+			fill_type_cmp_fmgr_info(&interval_type_cmp, value_type, prel->atttype);
 
 			if (SPI_connect() != SPI_OK_CONNECT)
 				elog(ERROR, "Could not connect using SPI");

src/pl_funcs.c

@@ -12,6 +12,7 @@
 #include "init.h"
 #include "utils.h"
 #include "relation_info.h"
+#include "xact_handling.h"
 
 #include "catalog/indexing.h"
 #include "commands/sequence.h"
@@ -48,6 +49,7 @@ PG_FUNCTION_INFO_V1( build_update_trigger_name );
 PG_FUNCTION_INFO_V1( is_date_type );
 PG_FUNCTION_INFO_V1( is_attribute_nullable );
 PG_FUNCTION_INFO_V1( add_to_pathman_config );
+PG_FUNCTION_INFO_V1( lock_partitioned_relation );
 PG_FUNCTION_INFO_V1( debug_capture );
 
 
@@ -211,29 +213,10 @@ find_or_create_range_partition(PG_FUNCTION_ARGS)
 		PG_RETURN_NULL();
 	else
 	{
-		Oid child_oid = InvalidOid;
-
-		/* FIXME: useless double-checked lock (no new data) */
-		LWLockAcquire(pmstate->load_config_lock, LW_EXCLUSIVE);
-		LWLockAcquire(pmstate->edit_partitions_lock, LW_EXCLUSIVE);
-
-		/*
-		 * Check if someone else has already created partition.
-		 */
-		search_state = search_range_partition_eq(value, &cmp_func, prel,
-												 &found_rentry);
-		if (search_state == SEARCH_RANGEREL_FOUND)
-		{
-			LWLockRelease(pmstate->load_config_lock);
-			LWLockRelease(pmstate->edit_partitions_lock);
-
-			PG_RETURN_OID(found_rentry.child_oid);
-		}
-
-		child_oid = create_partitions(parent_oid, value, value_type);
+		Oid	child_oid = create_partitions(parent_oid, value, value_type);
 
-		LWLockRelease(pmstate->load_config_lock);
-		LWLockRelease(pmstate->edit_partitions_lock);
+		/* get_pathman_relation_info() will refresh this entry */
+		invalidate_pathman_relation_info(parent_oid, NULL);
 
 		PG_RETURN_OID(child_oid);
 	}
@@ -680,6 +663,21 @@ add_to_pathman_config(PG_FUNCTION_ARGS)
 }
 
 
+/*
+ * Acquire appropriate lock on a partitioned relation.
+ */
+Datum
+lock_partitioned_relation(PG_FUNCTION_ARGS)
+{
+	Oid			relid = PG_GETARG_OID(0);
+
+	/* Lock partitioned relation till transaction's end */
+	xact_lock_partitioned_rel(relid);
+
+	PG_RETURN_VOID();
+}
+
+
 /*
  * NOTE: used for DEBUG, set breakpoint here.
  */

src/relation_info.c

@@ -11,6 +11,7 @@
 #include "relation_info.h"
 #include "init.h"
 #include "utils.h"
+#include "xact_handling.h"
 
 #include "access/htup_details.h"
 #include "access/xact.h"
@@ -240,6 +241,22 @@ get_pathman_relation_info(Oid relid)
 	return prel;
 }
 
+/* Acquire lock on a table and try to get PartRelationInfo */
+const PartRelationInfo *
+get_pathman_relation_info_after_lock(Oid relid, bool unlock_if_not_found)
+{
+	const PartRelationInfo *prel;
+
+	/* Restrict concurrent partition creation (it's dangerous) */
+	xact_lock_partitioned_rel(relid);
+
+	prel = get_pathman_relation_info(relid);
+	if (!prel && unlock_if_not_found)
+		xact_unlock_partitioned_rel(relid);
+
+	return prel;
+}
+
 /* Remove PartRelationInfo from local cache. */
 void
 remove_pathman_relation_info(Oid relid)

src/relation_info.h

@@ -121,6 +121,8 @@ const PartRelationInfo *refresh_pathman_relation_info(Oid relid,
 void invalidate_pathman_relation_info(Oid relid, bool *found);
 void remove_pathman_relation_info(Oid relid);
 const PartRelationInfo *get_pathman_relation_info(Oid relid);
+const PartRelationInfo *get_pathman_relation_info_after_lock(Oid relid,
+															 bool unlock_if_not_found);
 
 void delay_pathman_shutdown(void);
 void delay_invalidation_parent_rel(Oid parent);