check transaction isolation level for blocking partitioning operations

Author: funbringer

init.sql

@@ -655,7 +655,6 @@ LANGUAGE C STRICT;
  RETURNS VOID AS 'pg_pathman', 'lock_partitioned_relation'
  LANGUAGE C STRICT;
 
-
 /*
  * Lock relation to restrict concurrent modification of data.
  */
@@ -664,6 +663,15 @@ LANGUAGE C STRICT;
  RETURNS VOID AS 'pg_pathman', 'lock_relation_modification'
  LANGUAGE C STRICT;
 
+/*
+ * Check if we can distribute data without bad consequences.
+ */
+ CREATE OR REPLACE FUNCTION @extschema@.common_blocking_partitioning_checks(
+	 REGCLASS)
+ RETURNS VOID AS 'pg_pathman', 'common_blocking_partitioning_checks'
+ LANGUAGE C STRICT;
+
+
 /*
  * DEBUG: Place this inside some plpgsql fuction and set breakpoint.
  */

range.sql

@@ -98,7 +98,10 @@ BEGIN
 	PERFORM @extschema@.lock_partitioned_relation(parent_relid);
 
 	IF partition_data = true THEN
-		/* Acquire data modification lock */
+		/* Perform some checks regarding the blocking partitioning */
+		PERFORM @extschema@.common_blocking_partitioning_checks(parent_relid);
+
+		/* Acquire data modification lock (prevent further modifications) */
 		PERFORM @extschema@.lock_relation_modification(parent_relid);
 	END IF;
 
@@ -200,7 +203,10 @@ BEGIN
 	PERFORM @extschema@.lock_partitioned_relation(parent_relid);
 
 	IF partition_data = true THEN
-		/* Acquire data modification lock */
+		/* Perform some checks regarding the blocking partitioning */
+		PERFORM @extschema@.common_blocking_partitioning_checks(parent_relid);
+
+		/* Acquire data modification lock (prevent further modifications) */
 		PERFORM @extschema@.lock_relation_modification(parent_relid);
 	END IF;
 
@@ -300,7 +306,10 @@ BEGIN
 	PERFORM @extschema@.lock_partitioned_relation(parent_relid);
 
 	IF partition_data = true THEN
-		/* Acquire data modification lock */
+		/* Perform some checks regarding the blocking partitioning */
+		PERFORM @extschema@.common_blocking_partitioning_checks(parent_relid);
+
+		/* Acquire data modification lock (prevent further modifications) */
 		PERFORM @extschema@.lock_relation_modification(parent_relid);
 	END IF;
 
@@ -373,7 +382,10 @@ BEGIN
 	PERFORM @extschema@.lock_partitioned_relation(parent_relid);
 
 	IF partition_data = true THEN
-		/* Acquire data modification lock */
+		/* Perform some checks regarding the blocking partitioning */
+		PERFORM @extschema@.common_blocking_partitioning_checks(parent_relid);
+
+		/* Acquire data modification lock (prevent further modifications) */
 		PERFORM @extschema@.lock_relation_modification(parent_relid);
 	END IF;
 
@@ -521,7 +533,8 @@ BEGIN
 	/* Acquire exclusive lock on parent */
 	PERFORM @extschema@.lock_partitioned_relation(v_parent_relid);
 
-	/* Acquire data modification lock */
+	/* Acquire data modification lock (prevent further modifications) */
+	PERFORM @extschema@.common_blocking_partitioning_checks(p_partition);
 	PERFORM @extschema@.lock_relation_modification(p_partition);
 
 	SELECT attname, parttype
@@ -608,8 +621,10 @@ BEGIN
 	v_parent_relid1 := @extschema@.get_parent_of_partition(partition1);
 	v_parent_relid2 := @extschema@.get_parent_of_partition(partition2);
 
-	/* Acquire data modification lock */
+	/* Acquire data modification locks (prevent further modifications) */
+	PERFORM @extschema@.common_blocking_partitioning_checks(partition1);
 	PERFORM @extschema@.lock_relation_modification(partition1);
+	PERFORM @extschema@.common_blocking_partitioning_checks(partition2);
 	PERFORM @extschema@.lock_relation_modification(partition2);
 
 	IF v_parent_relid1 != v_parent_relid2 THEN

src/pathman_workers.c

@@ -361,7 +361,7 @@ bgw_main_spawn_partitions(Datum main_arg)
 #endif
 
 	/* Check again if there's a conflicting lock */
-	if (xact_conflicting_lock_exists(args->partitioned_table))
+	if (xact_bgw_conflicting_lock_exists(args->partitioned_table))
 	{
 		elog(LOG, "%s: there's a conflicting lock on relation \"%s\"",
 			 spawn_partitions_bgw,

src/pg_pathman.c

@@ -976,7 +976,7 @@ create_partitions(Oid relid, Datum value, Oid value_type)
 		 * If table has been partitioned in some previous xact AND
 		 * we don't hold any conflicting locks, run BGWorker.
 		 */
-		if (part_in_prev_xact && !xact_conflicting_lock_exists(relid))
+		if (part_in_prev_xact && !xact_bgw_conflicting_lock_exists(relid))
 		{
 			elog(DEBUG2, "create_partitions(): chose BGWorker [%u]", MyProcPid);
 			last_partition = create_partitions_bg_worker(relid, value, value_type);

src/pl_funcs.c

@@ -54,6 +54,7 @@ PG_FUNCTION_INFO_V1( add_to_pathman_config );
 PG_FUNCTION_INFO_V1( invalidate_relcache );
 PG_FUNCTION_INFO_V1( lock_partitioned_relation );
 PG_FUNCTION_INFO_V1( lock_relation_modification );
+PG_FUNCTION_INFO_V1( common_blocking_partitioning_checks );
 PG_FUNCTION_INFO_V1( debug_capture );
 
 
@@ -713,6 +714,25 @@ lock_relation_modification(PG_FUNCTION_ARGS)
 	PG_RETURN_VOID();
 }
 
+Datum
+common_blocking_partitioning_checks(PG_FUNCTION_ARGS)
+{
+	Oid			relid = PG_GETARG_OID(0);
+
+	if (!xact_is_level_read_committed())
+		ereport(ERROR,
+				(errmsg("Cannot perform blocking partitioning operation"),
+				 errdetail("Expected READ COMMITTED isolation level")));
+
+	if (xact_is_table_being_modified(relid))
+		ereport(ERROR,
+				(errmsg("Cannot perform blocking partitioning operation"),
+				 errdetail("Table \"%s\" is being modified concurrently",
+						   get_rel_name_or_relid(relid))));
+
+	PG_RETURN_VOID();
+}
+
 
 /*
  * NOTE: used for DEBUG, set breakpoint here.

src/xact_handling.c

@@ -11,6 +11,7 @@
 #include "xact_handling.h"
 
 #include "postgres.h"
+#include "access/xact.h"
 #include "catalog/catalog.h"
 #include "miscadmin.h"
 #include "storage/lmgr.h"
@@ -45,7 +46,7 @@ xact_unlock_partitioned_rel(Oid relid)
 void
 xact_lock_rel_data(Oid relid)
 {
-	LockRelationOid(relid, RowExclusiveLock);
+	LockRelationOid(relid, ShareLock);
 }
 
 /*
@@ -54,15 +55,15 @@ xact_lock_rel_data(Oid relid)
 void
 xact_unlock_rel_data(Oid relid)
 {
-	UnlockRelationOid(relid, RowExclusiveLock);
+	UnlockRelationOid(relid, ShareLock);
 }
 
 /*
  * Check whether we already hold a lock that
  * might conflict with partition spawning BGW.
  */
 bool
-xact_conflicting_lock_exists(Oid relid)
+xact_bgw_conflicting_lock_exists(Oid relid)
 {
 	LOCKMODE	lockmode;
 
@@ -78,6 +79,37 @@ xact_conflicting_lock_exists(Oid relid)
 	return false;
 }
 
+/*
+ * Check if table is being modified
+ * concurrently in a separate transaction.
+ */
+bool
+xact_is_table_being_modified(Oid relid)
+{
+	/*
+	 * Check if someone has already started a
+	 * transaction and modified table's contents.
+	 */
+	if (ConditionalLockRelationOid(relid, ExclusiveLock))
+	{
+		UnlockRelationOid(relid, ExclusiveLock);
+		return false;
+	}
+
+	return true;
+}
+
+/*
+ * Check if current transaction's level is READ COMMITTED.
+ */
+bool
+xact_is_level_read_committed(void)
+{
+	if (XactIsoLevel <= XACT_READ_COMMITTED)
+		return true;
+
+	return false;
+}
 
 /*
  * Do we hold the specified lock?

src/xact_handling.h

@@ -14,11 +14,6 @@
 #include "pathman.h"
 
 
-/*
- * List of partitioned relations locked by this backend (plain Oids).
- */
-extern List	   *locked_by_me;
-
 /*
  * Transaction locks.
  */
@@ -28,6 +23,11 @@ void xact_unlock_partitioned_rel(Oid relid);
 void xact_lock_rel_data(Oid relid);
 void xact_unlock_rel_data(Oid relid);
 
-bool xact_conflicting_lock_exists(Oid relid);
+/*
+ * Utility checks.
+ */
+bool xact_bgw_conflicting_lock_exists(Oid relid);
+bool xact_is_table_being_modified(Oid relid);
+bool xact_is_level_read_committed(void);
 
 #endif