Merge Dmitry's patch for split_range_partitions concurrency

Author: ildus

.gitignore

@@ -13,3 +13,4 @@ pg_pathman--*.sql
 tags
 cscope*
 Dockerfile
+testgres

src/hooks.c

@@ -706,7 +706,8 @@ pathman_shmem_startup_hook(void)
 void
 pathman_relcache_hook(Datum arg, Oid relid)
 {
-	Oid parent_relid;
+	Oid		parent_relid;
+	bool	parent_found;
 
 	/* Hooks can be disabled */
 	if (!pathman_hooks_enabled)
@@ -732,6 +733,9 @@ pathman_relcache_hook(Datum arg, Oid relid)
 	/* It *might have been a partition*, invalidate parent */
 	if (OidIsValid(parent_relid))
 	{
+		/* Invalidate PartRelationInfo cache if needed */
+		invalidate_pathman_relation_info(parent_relid, &parent_found);
+
 		delay_invalidation_parent_rel(parent_relid);
 
 		elog(DEBUG2, "Invalidation message for partition %u [%u]",

src/include/partition_filter.h

@@ -40,6 +40,7 @@ typedef struct
 	Oid					partid;				/* partition's relid */
 	ResultRelInfo	   *result_rel_info;	/* cached ResultRelInfo */
 	TupleConversionMap *tuple_map;			/* tuple conversion map (parent => child) */
+	bool				first_time;			/* did this entry exist? */
 } ResultRelInfoHolder;
 
 
@@ -133,6 +134,7 @@ Oid * find_partitions_for_value(Datum value, Oid value_type,
 								const PartRelationInfo *prel,
 								int *nparts);
 
+/* Transform partition's Oid into ResultRelInfo */
 ResultRelInfoHolder * select_partition_for_insert(Datum value, Oid value_type,
 												  const PartRelationInfo *prel,
 												  ResultPartsStorage *parts_storage,

src/partition_filter.c

@@ -19,6 +19,7 @@
 #include "catalog/pg_type.h"
 #include "foreign/fdwapi.h"
 #include "foreign/foreign.h"
+#include "miscadmin.h"
 #include "nodes/nodeFuncs.h"
 #include "rewrite/rewriteManip.h"
 #include "utils/guc.h"
@@ -239,6 +240,9 @@ scan_result_parts_storage(Oid partid, ResultPartsStorage *parts_storage)
 							 (const void *) &partid,
 							 HASH_ENTER, &found);
 
+	/* Don't execute recheck_partition_for_value() every time */
+	rri_holder->first_time = !found;
+
 	/* If not found, create & cache new ResultRelInfo */
 	if (!found)
 	{
@@ -432,43 +436,52 @@ select_partition_for_insert(Datum value, Oid value_type,
 
 	do
 	{
+		bool refresh_prel;
+
 		/* Search for matching partitions */
 		parts = find_partitions_for_value(value, value_type, prel, &nparts);
 
 		if (nparts > 1)
 			elog(ERROR, ERR_PART_ATTR_MULTIPLE);
 		else if (nparts == 0)
 		{
-			 selected_partid = create_partitions_for_value(parent_relid,
-														   value, value_type);
+			selected_partid = create_partitions_for_value(parent_relid,
+														  value, value_type);
 
-			 /* get_pathman_relation_info() will refresh this entry */
-			 invalidate_pathman_relation_info(parent_relid, NULL);
+			/* get_pathman_relation_info() will refresh this entry */
+			invalidate_pathman_relation_info(parent_relid, NULL);
 		}
 		else selected_partid = parts[0];
 
-		/* Replace parent table with a suitable partition */
+		pfree(parts);
+
+		/* Get ResultRelationInfo of a suitable partition (may invalidate 'prel') */
 		old_mcxt = MemoryContextSwitchTo(estate->es_query_cxt);
 		rri_holder = scan_result_parts_storage(selected_partid, parts_storage);
 		MemoryContextSwitchTo(old_mcxt);
 
-		/* This partition has been dropped, repeat with a new 'prel' */
-		if (rri_holder == NULL)
+		/*
+		 * We'd like to refresh PartRelationInfo if:
+		 *		1) Partition we've just chosen had been removed;
+		 *		2) It changed and we've chosen a visible partition for the 1st time.
+		 */
+		refresh_prel = !PointerIsValid(rri_holder) ||
+					  (!PrelIsValid(prel) && rri_holder->first_time && nparts > 0);
+
+		if (refresh_prel)
 		{
-			/* get_pathman_relation_info() will refresh this entry */
 			invalidate_pathman_relation_info(parent_relid, NULL);
 
 			/* Get a fresh PartRelationInfo */
 			prel = get_pathman_relation_info(parent_relid);
+			shout_if_prel_is_invalid(parent_relid, prel, PT_ANY);
 
-			/* Paranoid check (all partitions have vanished) */
-			if (!prel)
-				elog(ERROR, "table \"%s\" is not partitioned",
-					 get_rel_name_or_relid(parent_relid));
+			/* Reset selected partition */
+			rri_holder = NULL;
 		}
 	}
 	/* Loop until we get some result */
-	while (rri_holder == NULL);
+	while (!PointerIsValid(rri_holder));
 
 	return rri_holder;
 }

src/pg_pathman.c

@@ -778,6 +778,10 @@ walk_expr_tree(Expr *expr, const WalkerContext *context)
 {
 	WrapperNode *result = (WrapperNode *) palloc0(sizeof(WrapperNode));
 
+	Assert(PrelIsValid(context->prel));
+
+	CHECK_FOR_INTERRUPTS();
+
 	switch (nodeTag(expr))
 	{
 		/* Useful for INSERT optimization */
@@ -897,6 +901,8 @@ handle_const(const Const *c,
 {
 	const PartRelationInfo *prel = context->prel;
 
+	CHECK_FOR_INTERRUPTS();
+
 	/* Deal with missing strategy */
 	if (strategy == 0)
 		goto handle_const_return;
@@ -957,7 +963,7 @@ handle_const(const Const *c,
 											  &cast_success);
 
 					if (!cast_success)
-						elog(ERROR, "Cannot select partition: "
+						elog(ERROR, "cannot select partition: "
 									"unable to perform type cast");
 				}
 				/* Else use the Const's value */

src/pl_range_funcs.c

@@ -816,15 +816,15 @@ drop_range_partition_expand_next(PG_FUNCTION_ARGS)
 	/* Expand next partition if it exists */
 	if (i < PrelChildrenCount(prel) - 1)
 	{
-		RangeEntry	   *cur = &ranges[i],
-					   *next = &ranges[i + 1];
+		RangeEntry	   cur = ranges[i],
+					   next = ranges[i + 1];
 
 		/* Drop old constraint and create a new one */
-		modify_range_constraint(next->child_oid,
-								prel->expr_cstr,
+		modify_range_constraint(next.child_oid,
+								pstrdup(prel->expr_cstr),
 								prel->ev_type,
-								&cur->min,
-								&next->max);
+								&cur.min,
+								&next.max);
 	}
 
 	/* Finally drop this partition */

tests/python/partitioning_test.py

@@ -806,6 +806,81 @@ def con2_thread():
 		node.stop()
 		node.cleanup()
 
+	def test_conc_part_split_insert(self):
+		"""Test concurrent split_range_partition() + INSERT"""
+
+		# Create and start new instance
+		node = self.start_new_pathman_cluster(allows_streaming=False)
+
+		# Create table 'ins_test' and partition it
+		with node.connect() as con0:
+			con0.begin()
+			con0.execute('create table ins_test(val int not null)')
+			con0.execute("select create_range_partitions('ins_test', 'val', 1, 10, 10)")
+			con0.commit()
+
+		# Create two separate connections for this test
+		with node.connect() as con1, node.connect() as con2:
+
+			# Thread for connection #2 (it has to wait)
+			def con2_thread():
+				con2.begin()
+				con2.execute('insert into ins_test values (3), (6)')
+				con2.commit()
+
+			# Step 1: initilize con1
+			con1.begin()
+			con1.execute('select count(*) from ins_test') # load pathman's cache
+
+			# Step 2: initilize con2
+			con2.begin()
+			con2.execute('select count(*) from ins_test') # load pathman's cache
+			con2.commit() # unlock relations
+
+			# Step 3: split 'ins_test1' in con1 (success)
+			con1.execute("select split_range_partition('ins_test_1', 5)")
+
+			# Step 4: try inserting new values in con2 (waiting)
+			t = threading.Thread(target=con2_thread)
+			t.start()
+
+			# Step 5: wait until 't' locks
+			while True:
+				with node.connect() as con0:
+					locks = con0.execute("""
+						select count(*) from pg_locks where granted = 'f'
+					""")
+
+					if int(locks[0][0]) > 0:
+						break
+
+			# Step 6: finish split in con1 (success, unlock)
+			con1.commit()
+
+			# Step 7: wait for con2
+			t.join()
+
+			rows = con1.execute("""
+				select *, tableoid::regclass::text
+				from ins_test
+				order by val asc
+			""")
+
+			# check number of rows in table
+			self.assertEqual(len(rows), 2)
+
+			# check values that have been inserted
+			self.assertEqual(int(rows[0][0]), 3)
+			self.assertEqual(int(rows[1][0]), 6)
+
+			# check partitions that were chosen for insert
+			self.assertEqual(str(rows[0][1]), 'ins_test_1')
+			self.assertEqual(str(rows[1][1]), 'ins_test_11')
+
+		# Stop instance and finish work
+		node.stop()
+		node.cleanup()
+
 	def test_pg_dump(self):
 		"""
 		Test using dump and restore of partitioned table through pg_dump and pg_restore tools.