Function for generating CREATE TABLE for existing table.

Also create_hash_partitions in progress.

Author: arssher

Makefile

@@ -3,7 +3,7 @@ EXTENSION = pg_shardman        # the extension name
 DATA = pg_shardman--0.0.1.sql
 
 MODULE_big = pg_shardman
-OBJS = src/pg_shardman.o src/pg_shardman_cleanup.o
+OBJS = src/pg_shardman.o src/udf.o
 
 PG_CPPFLAGS += -Isrc/include
 

pg_shardman--0.0.1.sql

@@ -49,7 +49,7 @@ CREATE TABLE local_meta (
 INSERT INTO @extschema@.local_meta VALUES ('node_id', NULL);
 
 -- available commands
-CREATE TYPE cmd AS ENUM ('add_node', 'rm_node');
+CREATE TYPE cmd AS ENUM ('add_node', 'rm_node', 'create_hash_partitions');
 -- command status
 CREATE TYPE cmd_status AS ENUM ('waiting', 'canceled', 'failed', 'in progress', 'success');
 
@@ -213,6 +213,8 @@ CREATE FUNCTION set_node_id(node_id int) RETURNS void AS $$
 	UPDATE @extschema@.local_meta SET v = node_id WHERE k = 'node_id';
 $$ LANGUAGE sql;
 
+CREATE FUNCTION gen_create_table_sql(relation text) RETURNS text
+    AS 'pg_shardman' LANGUAGE C;
 
 -- Interface functions
 
@@ -237,3 +239,20 @@ BEGIN
 	INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, node_id);
 END
 $$ LANGUAGE plpgsql;
+
+-- Shard table with hash partitions. Params as in pathman, except for relation
+-- (master doesn't know oid of the table)
+CREATE FUNCTION create_hash_partitions(
+	node_id int, expr text, relation text, partitions_count int)
+	RETURNS void AS $$
+DECLARE
+	c_id int;
+BEGIN
+	INSERT INTO @extschema@.cmd_log VALUES (DEFAULT, 'create_hash_partitions')
+										   RETURNING id INTO c_id;
+	INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, node_id);
+	INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, expr);
+	INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, relation);
+	INSERT INTO @extschema@.cmd_opts VALUES (DEFAULT, c_id, partitions_count);
+END
+$$ LANGUAGE plpgsql;

postgresql.conf.master.template

@@ -1,5 +1,5 @@
 log_min_messages = INFO
-client_min_messages = LOG
+client_min_messages = NOTICE
 
 # for logical repl
 wal_level = logical

postgresql.conf.worker.template

@@ -1,7 +1,7 @@
 shared_preload_libraries = 'pg_pathman, pg_shardman'
 
 log_min_messages = INFO
-client_min_messages = LOG
+client_min_messages = NOTICE
 
 wal_level = logical
 

src/pg_shardman.c

@@ -55,7 +55,8 @@ static int insert_node(const char *connstring, int64 cmd_id);
 static bool node_in_cluster(int id);
 
 static void rm_node(Cmd *cmd);
-static bool is_node_active(int node_id);
+
+static void create_hash_partitions(Cmd *cmd);
 
 /* flags set by signal handlers */
 static volatile sig_atomic_t got_sigterm = false;
@@ -444,7 +445,7 @@ add_node(Cmd *cmd)
 	int node_id;
 	char *sql;
 
-	shmn_elog(LOG, "Adding node %s", connstring);
+	shmn_elog(INFO, "Adding node %s", connstring);
 	/* Try to execute command indefinitely until it succeeded or canceled */
 	while (!got_sigusr1 && !got_sigterm)
 	{
@@ -638,6 +639,7 @@ rm_node(Cmd *cmd)
 	int node_id = atoi(cmd->opts[0]);
 	char *sql;
 
+	elog(INFO, "Removing node %d ", node_id);
 	if (!node_in_cluster(node_id))
 	{
 		shmn_elog(WARNING, "node %d not in cluster, won't rm it.", node_id);
@@ -672,3 +674,21 @@ rm_node(Cmd *cmd)
 	pfree(sql);
 	elog(INFO, "Node %d successfully removed", node_id);
 }
+
+
+/*
+ * Partition table and get sql to create it;
+ * Add records about new table and partitions
+ */
+static void create_hash_partitions(Cmd *cmd)
+{
+	int node_id = atoi(cmd->opts[0]);
+	const char *expr = cmd->opts[1];
+	const char *relation = cmd->opts[2];
+	const char *partitions_count = atoi(cmd->opts[3]);
+	PGconn *conn = NULL;
+	PGresult *res = NULL;
+	char *sql;
+
+	shmn_elog(INFO, "Sharding table %s on node %d", relation, node_id);
+}

src/pg_shardman_cleanup.c

@@ -0,0 +1,109 @@
+#include "postgres.h"
+#include "commands/event_trigger.h"
+#include "executor/spi.h"
+#include "utils/builtins.h"
+#include "commands/dbcommands.h"
+#include "miscadmin.h"
+
+/*
+ * Must be called iff we are dropping extension. Checks that we are dropping
+ * pg_shardman extension and calls pg_shardman_cleanup to perform the actual
+ * cleanup.
+ */
+PG_FUNCTION_INFO_V1(pg_shardman_cleanup_c);
+Datum
+pg_shardman_cleanup_c(PG_FUNCTION_ARGS)
+{
+    EventTriggerData *trigdata;
+	DropStmt *stmt;
+	Value *value;
+	char *ext_name;
+
+    if (!CALLED_AS_EVENT_TRIGGER(fcinfo))  /* internal error */
+        elog(ERROR, "not fired by event trigger manager");
+
+    trigdata = (EventTriggerData *) fcinfo->context;
+	Assert(trigdata->parsetree->type == T_DropStmt);
+	stmt = (DropStmt *) trigdata->parsetree;
+	Assert(stmt->removeType == OBJECT_EXTENSION);
+	/* So it is list of pointers to Nodes */
+	Assert(stmt->objects->type == T_List);
+	/* To Value nodes, actually */
+	Assert(((Node *) (linitial(stmt->objects)))->type == T_String);
+	/* Seems like no way to use smth like linitial_node, because Value node
+	 * can have different types	*/
+
+	value = (Value *) linitial(stmt->objects);
+	ext_name = strVal(value);
+	if (strcmp(ext_name, "pg_shardman") == 0)
+	{
+		/* So we are dropping pg_shardman */
+		const char *cmd_sql = "select shardman.pg_shardman_cleanup();";
+
+		SPI_connect();
+		if (SPI_execute(cmd_sql, true, 0) < 0)
+			elog(FATAL, "Stmt failed: %s", cmd_sql);
+		SPI_finish();
+	}
+
+    PG_RETURN_NULL();
+}
+
+/*
+ * Generate CREATE TABLE sql for relation via pg_dump. Parameter is not
+ * REGCLASS because pg_dump can't handle oids anyway.
+ */
+PG_FUNCTION_INFO_V1(gen_create_table_sql);
+Datum
+gen_create_table_sql(PG_FUNCTION_ARGS)
+{
+	char pg_dump_path[MAXPGPATH];
+	Name db; /* current db name */
+	/* let the mmgr free that */
+	char *relation = text_to_cstring(PG_GETARG_TEXT_PP(0));
+	const size_t chunksize = 5; /* read max that bytes at time */
+	/* how much already allocated *including header* */
+	size_t pallocated = VARHDRSZ + chunksize;
+	text *sql = (text *) palloc(pallocated);
+	char *ptr = VARDATA(sql); /* ptr to first free byte */
+	char *cmd;
+	FILE *fp;
+	size_t bytes_read;
+	SET_VARSIZE(sql, VARHDRSZ);
+
+	/* find pg_dump location */
+	if (find_my_exec("pg_dump", pg_dump_path) != 0)
+	{
+		elog(ERROR, "Failed to find pg_dump location");
+	}
+	/* find current db */
+	db = (Name) palloc(NAMEDATALEN);
+	namestrcpy(db, get_database_name(MyDatabaseId));
+
+	cmd = psprintf("%s -t '%s' --schema-only '%s' 2>&1",
+				   pg_dump_path, relation, (NameStr(*db)));
+
+	if ((fp = popen(cmd, "r")) == NULL)
+	{
+		elog(ERROR, "Failed to run pg_dump, cmd %s", cmd);
+	}
+
+	while ((bytes_read = fread(ptr, sizeof(char), chunksize, fp)) != 0)
+	{
+		SET_VARSIZE(sql, VARSIZE_ANY(sql) + bytes_read);
+		if (pallocated - VARSIZE_ANY(sql) < chunksize)
+		{
+			pallocated *= 2;
+			sql = (text *) repalloc(sql, pallocated);
+		}
+		/* since we realloc, can't just += bytes_read here */
+		ptr = VARDATA(sql) + VARSIZE_ANY_EXHDR(sql);
+	}
+
+	if (pclose(fp))	{
+		elog(ERROR, "pg_dump exited with error status, output was\n%scmd was \n%s",
+			 text_to_cstring(sql), cmd);
+	}
+
+	PG_RETURN_TEXT_P(sql);
+}