clean init.sql up, improve function partition_table_concurrently()

funbringer · funbringer · commit 481008208503 · 2016-10-09T22:17:04.000+03:00
diff --git a/README.md b/README.md
@@ -124,9 +124,11 @@ Performs RANGE-partitioning from specified range for `relation` by partitioning
 ### Data migration
 
 ```plpgsql
-partition_table_concurrently(relation REGCLASS)
+partition_table_concurrently(relation   REGCLASS,
+                             batch_size INTEGER DEFAULT 1000,
+                             sleep_time FLOAT8 DEFAULT 1.0)
 ```
-Starts a background worker to move data from parent table to partitions. The worker utilizes short transactions to copy small batches of data (up to 10K rows per transaction) and thus doesn't significantly interfere with user's activity.
+Starts a background worker to move data from parent table to partitions. The worker utilizes short transactions to copy small batches of data (up to 10K rows per transaction) and thus doesn't significantly interfere with user's activity. If the worker is unable to lock rows of a batch, it sleeps for `sleep_time` seconds up to 60 times before the next attempt, and quits if it's still unable to lock the batch.
 
 ```plpgsql
 stop_concurrent_part_task(relation REGCLASS)
diff --git a/expected/pathman_basic.out b/expected/pathman_basic.out
@@ -1409,7 +1409,7 @@ CREATE TABLE replies(id SERIAL PRIMARY KEY, message_id INTEGER REFERENCES messag
 INSERT INTO messages SELECT g, md5(g::text) FROM generate_series(1, 10) as g;
 INSERT INTO replies SELECT g, g, md5(g::text) FROM generate_series(1, 10) as g;
 SELECT create_range_partitions('messages', 'id', 1, 100, 2);
-WARNING:  foreign key 'replies_message_id_fkey' references relation 'messages'
+WARNING:  foreign key "replies_message_id_fkey" references relation "messages"
 ERROR:  relation "messages" is referenced from other relations
 ALTER TABLE replies DROP CONSTRAINT replies_message_id_fkey;
 SELECT create_range_partitions('messages', 'id', 1, 100, 2);
diff --git a/init.sql b/init.sql
@@ -225,21 +225,27 @@ GRANT SELECT ON @extschema@.pathman_concurrent_part_tasks TO PUBLIC;
 /*
  * Partition table using ConcurrentPartWorker.
  */
-CREATE OR REPLACE FUNCTION @extschema@.partition_table_concurrently(relation regclass)
-RETURNS VOID AS 'pg_pathman', 'partition_table_concurrently' LANGUAGE C STRICT;
+CREATE OR REPLACE FUNCTION @extschema@.partition_table_concurrently(
+	relation		REGCLASS,
+	batch_size		INTEGER DEFAULT 1000,
+	sleep_time		FLOAT8 DEFAULT 1.0)
+RETURNS VOID AS 'pg_pathman', 'partition_table_concurrently'
+LANGUAGE C STRICT;
 
 /*
  * Stop concurrent partitioning task.
  */
-CREATE OR REPLACE FUNCTION @extschema@.stop_concurrent_part_task(relation regclass)
-RETURNS BOOL AS 'pg_pathman', 'stop_concurrent_part_task' LANGUAGE C STRICT;
+CREATE OR REPLACE FUNCTION @extschema@.stop_concurrent_part_task(
+	relation		REGCLASS)
+RETURNS BOOL AS 'pg_pathman', 'stop_concurrent_part_task'
+LANGUAGE C STRICT;
 
 
 /*
  * Copy rows to partitions concurrently.
  */
 CREATE OR REPLACE FUNCTION @extschema@._partition_data_concurrent(
-	p_relation		REGCLASS,
+	relation		REGCLASS,
 	p_min			ANYELEMENT DEFAULT NULL::text,
 	p_max			ANYELEMENT DEFAULT NULL::text,
 	p_limit			INT DEFAULT NULL,
@@ -254,7 +260,7 @@ DECLARE
 
 BEGIN
 	SELECT attname INTO v_attr
-	FROM @extschema@.pathman_config WHERE partrel = p_relation;
+	FROM @extschema@.pathman_config WHERE partrel = relation;
 
 	p_total := 0;
 
@@ -282,15 +288,15 @@ BEGIN
 	/* Lock rows and copy data */
 	RAISE NOTICE 'Copying data to partitions...';
 	EXECUTE format('SELECT array(SELECT ctid FROM ONLY %1$s %2$s %3$s FOR UPDATE NOWAIT)',
-				   p_relation, v_where_clause, v_limit_clause)
+				   relation, v_where_clause, v_limit_clause)
 	USING p_min, p_max
 	INTO ctids;
 
 	EXECUTE format('
 		WITH data AS (
 			DELETE FROM ONLY %1$s WHERE ctid = ANY($1) RETURNING *)
 		INSERT INTO %1$s SELECT * FROM data',
-		p_relation)
+		relation)
 	USING ctids;
 
 	/* Get number of inserted rows */
@@ -376,7 +382,7 @@ LANGUAGE plpgsql;
  * Suitable for every partitioning type.
  */
 CREATE OR REPLACE FUNCTION @extschema@.common_relation_checks(
-	p_relation		REGCLASS,
+	relation		REGCLASS,
 	p_attribute		TEXT)
 RETURNS BOOLEAN AS
 $$
@@ -388,33 +394,33 @@ DECLARE
 BEGIN
 	/* Ignore temporary tables */
 	SELECT relpersistence FROM pg_catalog.pg_class
-	WHERE oid = p_relation INTO rel_persistence;
+	WHERE oid = relation INTO rel_persistence;
 
 	IF rel_persistence = 't'::CHAR THEN
 		RAISE EXCEPTION 'temporary table "%" cannot be partitioned',
-						p_relation::TEXT;
+						relation::TEXT;
 	END IF;
 
 	IF EXISTS (SELECT * FROM @extschema@.pathman_config
-			   WHERE partrel = p_relation) THEN
-		RAISE EXCEPTION 'relation "%" has already been partitioned', p_relation;
+			   WHERE partrel = relation) THEN
+		RAISE EXCEPTION 'relation "%" has already been partitioned', relation;
 	END IF;
 
-	IF @extschema@.is_attribute_nullable(p_relation, p_attribute) THEN
+	IF @extschema@.is_attribute_nullable(relation, p_attribute) THEN
 		RAISE EXCEPTION 'partitioning key ''%'' must be NOT NULL', p_attribute;
 	END IF;
 
 	/* Check if there are foreign keys that reference the relation */
-	FOR v_rec IN (SELECT *
-				  FROM pg_constraint WHERE confrelid = p_relation::regclass::oid)
+	FOR v_rec IN (SELECT * FROM pg_catalog.pg_constraint
+				  WHERE confrelid = relation::REGCLASS::OID)
 	LOOP
 		is_referenced := TRUE;
-		RAISE WARNING 'foreign key ''%'' references relation ''%''',
-				v_rec.conname, p_relation;
+		RAISE WARNING 'foreign key "%" references relation "%"',
+				v_rec.conname, relation;
 	END LOOP;
 
 	IF is_referenced THEN
-		RAISE EXCEPTION 'relation "%" is referenced from other relations', p_relation;
+		RAISE EXCEPTION 'relation "%" is referenced from other relations', relation;
 	END IF;
 
 	RETURN TRUE;
@@ -627,7 +633,7 @@ BEGIN
 	LOOP
 		EXECUTE format('ALTER TABLE %s ADD %s',
 					   partition::TEXT,
-					   pg_get_constraintdef(rec.conid));
+					   pg_catalog.pg_get_constraintdef(rec.conid));
 	END LOOP;
 END
 $$ LANGUAGE plpgsql STRICT;
@@ -743,7 +749,8 @@ RETURNS BOOLEAN AS 'pg_pathman', 'add_to_pathman_config'
 LANGUAGE C;
 
 CREATE OR REPLACE FUNCTION @extschema@.invalidate_relcache(relid OID)
-RETURNS VOID AS 'pg_pathman' LANGUAGE C STRICT;
+RETURNS VOID AS 'pg_pathman'
+LANGUAGE C STRICT;
 
 
 /*
diff --git a/src/pathman_workers.c b/src/pathman_workers.c
@@ -574,9 +574,20 @@ Datum
 partition_table_concurrently(PG_FUNCTION_ARGS)
 {
 	Oid		relid = PG_GETARG_OID(0);
+	int32	batch_size = PG_GETARG_INT32(1);
+	float8	sleep_time = PG_GETARG_FLOAT8(2);
 	int		empty_slot_idx = -1,		/* do we have a slot for BGWorker? */
 			i;
 
+	/* Check batch_size */
+	if (batch_size < 1 || batch_size > 10000)
+		elog(ERROR, "\"batch_size\" should not be less than 1 "
+					"or greater than 10000");
+
+	/* Check sleep_time */
+	if (sleep_time < 0.5)
+		elog(ERROR, "\"sleep_time\" should not be less than 0.5");
+
 	/* Check if relation is a partitioned table */
 	shout_if_prel_is_invalid(relid,
 							 /* We also lock the parent relation */
@@ -631,8 +642,8 @@ partition_table_concurrently(PG_FUNCTION_ARGS)
 	{
 		/* Initialize concurrent part slot */
 		InitConcurrentPartSlot(&concurrent_part_slots[empty_slot_idx],
-							   GetUserId(), CPS_WORKING,
-							   MyDatabaseId, relid, 1000, 1.0);
+							   GetUserId(), CPS_WORKING, MyDatabaseId,
+							   relid, batch_size, sleep_time);
 
 		/* Now we can safely unlock slot for new BGWorker */
 		SpinLockRelease(&concurrent_part_slots[empty_slot_idx].mutex);
