Merge branch 'merge_concurrent' of github.com:postgrespro/pg_pathman into merge_concurrent_locks

Author: zilder

Diff for: src/hooks.c

@@ -521,6 +521,7 @@ pathman_relcache_hook(Datum arg, Oid relid)
 		case PPS_ENTRY_NOT_FOUND:
 			{
 				/* NOTE: Remove NOT_USED when it's time */
+				delay_invalidation_parent_rel(partitioned_table);
 #ifdef NOT_USED
 				elog(DEBUG2, "Invalidation message for relation %u [%u]",
 					 relid, MyProcPid);

Diff for: src/init.c

@@ -118,6 +118,17 @@ init_main_pathman_toggle(void)
 							 NULL,
 							 pg_pathman_enable_assign_hook,
 							 NULL);
+
+	DefineCustomBoolVariable("pg_pathman.enable_auto_partition",
+							 "Enables auto partition propagation",
+							 NULL,
+							 &pg_pathman_init_state.auto_partition,
+							 true,
+							 PGC_USERSET,
+							 0,
+							 NULL,
+							 NULL,
+							 NULL);
 }
 
 /*

Diff for: src/init.h

@@ -26,6 +26,7 @@
 typedef struct
 {
 	bool 	pg_pathman_enable;		/* GUC variable implementation */
+	bool	auto_partition;			/* GUC variable for auto partition propagation */
 	bool	initialization_needed;	/* do we need to perform init? */
 } PathmanInitState;
 
@@ -52,6 +53,21 @@ extern PathmanInitState 	pg_pathman_init_state;
  */
 #define IsPathmanReady()		( IsPathmanInitialized() && IsPathmanEnabled() )
 
+/*
+ * Check if auto partition propagation enabled
+ */
+#define IsAutoPartitionEnabled() ( pg_pathman_init_state.auto_partition )
+
+/*
+ * Enable/disable auto partition propagation. Note that this only works if
+ * partitioned relation supports this. See enable_auto() and disable_auto()
+ * functions.
+ */
+#define SetAutoPartitionEnabled(value) \
+	do { \
+		pg_pathman_init_state.auto_partition = value; \
+	} while (0)
+
 /*
  * Emergency disable mechanism.
  */

Diff for: src/partition_filter.c

@@ -11,6 +11,7 @@
 #include "partition_filter.h"
 #include "nodes_common.h"
 #include "utils.h"
+#include "init.h"
 
 #include "utils/guc.h"
 #include "utils/memutils.h"
@@ -204,7 +205,11 @@ partition_filter_exec(CustomScanState *node)
 			elog(ERROR, "PartitionFilter selected more than one partition");
 		else if (nparts == 0)
 		{
-			if (prel->auto_partition)
+			/*
+			 * If auto partition propagation is enabled then try to create
+			 * new partitions for the key
+			 */
+			if (prel->auto_partition && IsAutoPartitionEnabled())
 			{
 				selected_partid = create_partitions(state->partitioned_table,
 													state->temp_const.constvalue,

Diff for: src/pathman_workers.c

@@ -417,6 +417,9 @@ bgw_main_concurrent_part(Datum main_arg)
 	part_slot = &concurrent_part_slots[DatumGetInt32(main_arg)];
 	part_slot->pid = MyProcPid;
 
+	/* Disable auto partition propagation */
+	SetAutoPartitionEnabled(false);
+
 	/* Establish connection and start transaction */
 	BackgroundWorkerInitializeConnectionByOid(part_slot->dbid, part_slot->userid);
 
@@ -494,10 +497,11 @@ bgw_main_concurrent_part(Datum main_arg)
 			/* Print messsage for this BGWorker to server log */
 			sleep_time_str = datum_to_cstring(Float8GetDatum(part_slot->sleep_time),
 											  FLOAT8OID);
+			failures_count++;
 			ereport(LOG,
 					(errmsg("%s: %s", concurrent_part_bgw, error->message),
 					 errdetail("Attempt: %d/%d, sleep time: %s",
-							   failures_count + 1,
+							   failures_count,
 							   PART_WORKER_MAX_ATTEMPTS,
 							   sleep_time_str)));
 			pfree(sleep_time_str); /* free the time string */
@@ -509,7 +513,7 @@ bgw_main_concurrent_part(Datum main_arg)
 			 * concurrent user queries. Check that attempts count doesn't exceed
 			 * some reasonable value
 			 */
-			if (failures_count++ >= PART_WORKER_MAX_ATTEMPTS)
+			if (failures_count >= PART_WORKER_MAX_ATTEMPTS)
 			{
 				/* Mark slot as FREE */
 				cps_set_status(part_slot, CPS_FREE);
@@ -593,7 +597,6 @@ partition_table_concurrently(PG_FUNCTION_ARGS)
 							 get_pathman_relation_info_after_lock(relid, true),
 							 /* Partitioning type does not matter here */
 							 PT_INDIFFERENT);
-
 	/*
 	 * Look for an empty slot and also check that a concurrent
 	 * partitioning operation for this table hasn't been started yet

Diff for: src/pg_pathman.c

@@ -446,43 +446,61 @@ append_child_relation(PlannerInfo *root, RelOptInfo *rel, Index rti,
 	memcpy(childrel->attr_widths, rel->attr_widths,
 		   (rel->max_attr - rel->min_attr + 1) * sizeof(int32));
 
-
-	/* Copy restrictions */
+	/*
+	 * Copy restrictions. If it's not the parent table then copy only those
+	 * restrictions that reference to this partition
+	 */
 	childrel->baserestrictinfo = NIL;
-	forboth(lc, wrappers, lc2, rel->baserestrictinfo)
+	if (rte->relid != childOid)
 	{
-		bool alwaysTrue;
-		WrapperNode *wrap = (WrapperNode *) lfirst(lc);
-		Node *new_clause = wrapper_make_expression(wrap, index, &alwaysTrue);
-		RestrictInfo *old_rinfo = (RestrictInfo *) lfirst(lc2);
-
-		if (alwaysTrue)
+		forboth(lc, wrappers, lc2, rel->baserestrictinfo)
 		{
-			continue;
-		}
-		Assert(new_clause);
+			bool alwaysTrue;
+			WrapperNode *wrap = (WrapperNode *) lfirst(lc);
+			Node *new_clause = wrapper_make_expression(wrap, index, &alwaysTrue);
+			RestrictInfo *old_rinfo = (RestrictInfo *) lfirst(lc2);
 
-		if (and_clause((Node *) new_clause))
-		{
-			ListCell *alc;
+			if (alwaysTrue)
+			{
+				continue;
+			}
+			Assert(new_clause);
 
-			foreach(alc, ((BoolExpr *) new_clause)->args)
+			if (and_clause((Node *) new_clause))
 			{
-				Node *arg = (Node *) lfirst(alc);
-				RestrictInfo *new_rinfo = rebuild_restrictinfo(arg, old_rinfo);
+				ListCell *alc;
+
+				foreach(alc, ((BoolExpr *) new_clause)->args)
+				{
+					Node *arg = (Node *) lfirst(alc);
+					RestrictInfo *new_rinfo = rebuild_restrictinfo(arg, old_rinfo);
 
+					change_varnos((Node *)new_rinfo, rel->relid, childrel->relid);
+					childrel->baserestrictinfo = lappend(childrel->baserestrictinfo,
+														 new_rinfo);
+				}
+			}
+			else
+			{
+				RestrictInfo *new_rinfo = rebuild_restrictinfo(new_clause, old_rinfo);
+
+				/* Replace old relids with new ones */
 				change_varnos((Node *)new_rinfo, rel->relid, childrel->relid);
+
 				childrel->baserestrictinfo = lappend(childrel->baserestrictinfo,
-													 new_rinfo);
+													 (void *) new_rinfo);
 			}
 		}
-		else
+	}
+	/* If it's the parent table then copy all restrictions */
+	else
+	{
+		foreach(lc, rel->baserestrictinfo)
 		{
-			RestrictInfo *new_rinfo = rebuild_restrictinfo(new_clause, old_rinfo);
+			RestrictInfo *rinfo = (RestrictInfo *) lfirst(lc);
+			RestrictInfo *new_rinfo = (RestrictInfo *) copyObject(rinfo);
 
-			/* Replace old relids with new ones */
 			change_varnos((Node *)new_rinfo, rel->relid, childrel->relid);
-
 			childrel->baserestrictinfo = lappend(childrel->baserestrictinfo,
 												 (void *) new_rinfo);
 		}