movempart in progress, fixed unpleasant bug in timeutils

Author: arssher

pg_shardman--0.0.1.sql

@@ -301,17 +301,12 @@ BEGIN
 END;
 $$ LANGUAGE plpgsql;
 
--- Create logical pgoutput replication slot, if not exists
+-- Recreate logical pgoutput replication slot. Drops existing slot.
 CREATE FUNCTION create_repslot(slot_name text) RETURNS void AS $$
-DECLARE
-	slot_exists bool;
 BEGIN
-	EXECUTE format('SELECT EXISTS (SELECT * FROM pg_replication_slots
-				   WHERE slot_name=%L)', slot_name) INTO slot_exists;
-	IF NOT slot_exists THEN
-		EXECUTE format('SELECT pg_create_logical_replication_slot(%L, %L)',
-					   slot_name, 'pgoutput');
-	END IF;
+	PERFORM shardman.drop_repslot(slot_name);
+	EXECUTE format('SELECT pg_create_logical_replication_slot(%L, %L)',
+				   slot_name, 'pgoutput');
 END
 $$ LANGUAGE plpgsql;
 
@@ -321,7 +316,7 @@ DECLARE
 	slot_exists bool;
 BEGIN
 	EXECUTE format('SELECT EXISTS (SELECT * FROM pg_replication_slots
-				   WHERE slot_name=%L)', slot_name) INTO slot_exists;
+				   WHERE slot_name = %L)', slot_name) INTO slot_exists;
 	IF slot_exists THEN
 		EXECUTE format('SELECT pg_drop_replication_slot(%L)', slot_name);
 	END IF;

src/include/pg_shardman.h

@@ -43,5 +43,6 @@ extern void update_cmd_status(int64 id, const char *new_status);
 extern void cmd_canceled(Cmd *cmd);
 extern char *get_worker_node_connstr(int node_id);
 extern int32 get_partition_owner(const char *part_name);
+extern char *get_partition_relation(const char *part_name);
 
 #endif							/* PG_SHARDMAN_H */

src/pg_shardman.c

@@ -525,13 +525,13 @@ add_node(Cmd *cmd)
 		}
 		PQclear(res);
 
-		/* Create replication slot, if not yet */
+		/* Create replication slot */
 		sql = psprintf("select shardman.create_repslot('shardman_meta_sub_%d');",
 					   node_id);
 		void_spi(sql);
 		pfree(sql);
 
-		/* Create subscription and set node it on itself */
+		/* Create subscription and set node id on itself */
 		sql = psprintf(
 			"create subscription shardman_meta_sub connection '%s'"
 			"publication shardman_meta_pub with (create_slot = false,"
@@ -752,3 +752,41 @@ get_partition_owner(const char *part_name)
 	SPI_EPILOG;
 	return owner;
 }
+
+/*
+ * Get relation name of partition part_name. Memory is palloc'ed.
+ * NULL is returned, if there is no such partition.
+ */
+char*
+get_partition_relation(const char *part_name)
+{
+	MemoryContext oldcxt = CurrentMemoryContext;
+	char *sql = psprintf("select relation from shardman.partitions"
+						 " where part_name = '%s';", part_name);
+	char *res;
+
+	SPI_PROLOG;
+
+	if (SPI_execute(sql, true, 0) < 0)
+	{
+		shmn_elog(FATAL, "Stmt failed : %s", sql);
+	}
+	pfree(sql);
+
+	if (SPI_processed == 0)
+	{
+		res = NULL;
+	}
+	else
+	{
+		HeapTuple tuple = SPI_tuptable->vals[0];
+		TupleDesc rowdesc = SPI_tuptable->tupdesc;
+		/* We need to allocate connstring in our ctxt, not spi's */
+		MemoryContext spicxt = MemoryContextSwitchTo(oldcxt);
+		res = SPI_getvalue(tuple, rowdesc, 1);
+		MemoryContextSwitchTo(spicxt);
+	}
+
+	SPI_EPILOG;
+	return res;
+}