Implement autopartitioning of tablespaces

Author: knizhnik

src/backend/access/common/reloptions.c

@@ -164,6 +164,15 @@ static relopt_bool boolRelOpts[] =
 
 static relopt_int intRelOpts[] =
 {
+	{
+		{
+			"max_files",
+			"Limit number of files in tablespace",
+			RELOPT_KIND_TABLESPACE,
+			AccessExclusiveLock
+		},
+		0, 0, 10000 /* by default: no limit, restrict max value to 10k otherwise counting files in tablespace on each DDL will become too expensive  */
+	},
 	{
 		{
 			"fillfactor",
@@ -1690,7 +1699,8 @@ tablespace_reloptions(Datum reloptions, bool validate)
 	static const relopt_parse_elt tab[] = {
 		{"random_page_cost", RELOPT_TYPE_REAL, offsetof(TableSpaceOpts, random_page_cost)},
 		{"seq_page_cost", RELOPT_TYPE_REAL, offsetof(TableSpaceOpts, seq_page_cost)},
-		{"effective_io_concurrency", RELOPT_TYPE_INT, offsetof(TableSpaceOpts, effective_io_concurrency)}
+		{"effective_io_concurrency", RELOPT_TYPE_INT, offsetof(TableSpaceOpts, effective_io_concurrency)},
+		{"max_files", RELOPT_TYPE_INT, offsetof(TableSpaceOpts, max_files)}
 	};
 
 	return (bytea *) build_reloptions(reloptions, validate,

src/backend/commands/indexcmds.c

@@ -666,6 +666,8 @@ DefineIndex(Oid relationId,
 		if (aclresult != ACLCHECK_OK)
 			aclcheck_error(aclresult, OBJECT_TABLESPACE,
 						   get_tablespace_name(tablespaceId));
+
+		tablespaceId = GetCurrentTablespaceSegment(tablespaceId);
 	}
 
 	/*

src/backend/commands/tablecmds.c

@@ -528,7 +528,6 @@ static void refuseDupeIndexAttach(Relation parentIdx, Relation partIdx,
 static List *GetParentedForeignKeyRefs(Relation partition);
 static void ATDetachCheckNoForeignKeyRefs(Relation partition);
 
-
 /* ----------------------------------------------------------------
  *		DefineRelation
  *				Creates a new relation.
@@ -699,8 +698,9 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
 		if (aclresult != ACLCHECK_OK)
 			aclcheck_error(aclresult, OBJECT_TABLESPACE,
 						   get_tablespace_name(tablespaceId));
-	}
 
+		tablespaceId = GetCurrentTablespaceSegment(tablespaceId);
+	}
 	/* In all cases disallow placing user relations in pg_global */
 	if (tablespaceId == GLOBALTABLESPACE_OID)
 		ereport(ERROR,

src/backend/commands/tablespace.c

@@ -82,6 +82,7 @@
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
+#include "utils/spccache.h"
 #include "utils/varlena.h"
 
 /* GUC variables */
@@ -222,6 +223,54 @@ TablespaceCreateDbspace(Oid spcNode, Oid dbNode, bool isRedo)
 	pfree(dir);
 }
 
+static int
+getIntOption(List *options, char const *name, int defaultValue)
+{
+	ListCell   *cell;
+
+	foreach(cell, options)
+	{
+		DefElem    *def = (DefElem *) lfirst(cell);
+		if (strcmp(def->defname, name) == 0)
+		{
+			if (def->arg != NULL)
+			{
+				switch (nodeTag(def->arg))
+				{
+				  case T_Integer:
+					return (int32) intVal(def->arg);
+				  case T_String:
+					return atoi(strVal(def->arg));
+				  default:
+					ereport(ERROR,
+							(errcode(ERRCODE_SYNTAX_ERROR),
+							 errmsg("%s requires an integer value",
+									name)));
+				}
+				break;
+			}
+		}
+	}
+	return defaultValue;
+}
+
+static void
+AtomicChangeLink(char const* link_path, char const* new_location)
+{
+	char* tmp_link_path = psprintf("%s.tmp", link_path);
+	(void)unlink(tmp_link_path); /* remove temporary link if exists */
+	if (symlink(new_location, tmp_link_path) < 0)
+		ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not create symbolic link \"%s\": %m",
+							tmp_link_path)));
+	if (rename(tmp_link_path, link_path) < 0)
+		ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not rename symbolic link \"%s\": %m",
+							link_path)));
+}
+
 /*
  * Create a table space
  *
@@ -366,6 +415,30 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 	/* Post creation hook for new tablespace */
 	InvokeObjectPostCreateHook(TableSpaceRelationId, tablespaceoid, 0);
 
+	if (getIntOption(stmt->options, "max_files", 0) != 0)
+	{
+		/* Limited tablespace */
+		char* segment = psprintf("%s/%u", location, tablespaceoid);
+		char* link_path = psprintf("%s/current_segment", location);
+
+		pfree(location);
+		location = segment;
+		if (MakePGDirectory(location) < 0)
+		{
+			if (errno == EEXIST)
+				ereport(ERROR,
+						(errcode(ERRCODE_OBJECT_IN_USE),
+						 errmsg("directory \"%s\" already in use as a tablespace",
+								location)));
+			else
+				ereport(ERROR,
+						(errcode_for_file_access(),
+						 errmsg("could not create directory \"%s\": %m",
+								location)));
+		}
+		AtomicChangeLink(link_path, location);
+	}
+
 	create_tablespace_directories(location, tablespaceoid);
 
 	/* Record the filesystem change in XLOG */
@@ -1554,3 +1627,161 @@ tblspc_redo(XLogReaderState *record)
 	else
 		elog(PANIC, "tblspc_redo: unknown op code %u", info);
 }
+
+
+static int
+RecursiveCountFiles(char const* linkloc_with_version_dir, int level)
+{
+	char *subfile;
+	int count = 0;
+	DIR	 *dirdesc = AllocateDir(linkloc_with_version_dir);
+	struct dirent *de;
+
+	if (dirdesc == NULL)
+	{
+		return 0;
+	}
+	while ((de = ReadDir(dirdesc, linkloc_with_version_dir)) != NULL)
+	{
+		if (strcmp(de->d_name, ".") == 0 ||
+			strcmp(de->d_name, "..") == 0)
+			continue;
+
+		subfile = psprintf("%s/%s", linkloc_with_version_dir, de->d_name);
+		/* Assume that relations files are always at second level */
+		count += level > 0 ? RecursiveCountFiles(subfile, level - 1) : 1;
+	}
+	FreeDir(dirdesc);
+	return count;
+}
+
+static int
+CountTablespaceFiles(char const* curr_segment)
+{
+	char* linkloc_with_version_dir = psprintf("%s/%s", curr_segment, TABLESPACE_VERSION_DIRECTORY);
+	return RecursiveCountFiles(linkloc_with_version_dir, 1);
+}
+
+static Oid
+GetCurrentTablespaceSegmentId(char* segment_path)
+{
+	char* tblspc = strrchr(segment_path, '/');
+	if (tblspc == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("Invalid current tablespace segment \"%s\"",
+						segment_path)));
+	*tblspc++ = '\0';
+	return atoi(tblspc);
+}
+
+static char*
+GetCurrentTablespaceSegmentLocation(Oid tablespaceId)
+{
+	char link_target[MAXPGPATH];
+	char* link_path = psprintf("pg_tblspc/%u", tablespaceId);
+	int len = readlink(link_path, link_target, sizeof(link_target));
+	if (len < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read symbolic link \"%s\": %m",
+						link_path)));
+	link_target[len] = '\0';
+	pfree(link_path);
+	GetCurrentTablespaceSegmentId(link_target);
+	link_path = psprintf("%s/current_segment", link_target);
+	len = readlink(link_path, link_target, sizeof(link_target));
+	if (len < 0)
+		ereport(ERROR,
+				(errcode_for_file_access(),
+				 errmsg("could not read symbolic link \"%s\": %m",
+						link_path)));
+	link_target[len] = '\0';
+	pfree(link_path);
+	return pstrdup(link_target);
+}
+
+Oid
+GetCurrentTablespaceSegment(Oid tablespaceId)
+{
+	CreateTableSpaceStmt* stmt;
+	Relation	rel;
+	ScanKeyData entry[1];
+	TableScanDesc scan;
+	HeapTuple	tup;
+	Oid         spcowner;
+	bool		isnull;
+	List*       relopt;
+	int         num_files;
+	char*       orig_spcname;
+	char*       new_spcname;
+	char*       curr_segment;
+	bool        segment_is_used;
+	int         limit;
+	Oid         segmentId;
+
+	/* If there is limit for maximal files in tablespace, then count files in the current segment and create new one if needed */
+	limit = get_tablespace_max_files(tablespaceId);
+	if (limit == 0)
+		return tablespaceId;
+
+	curr_segment = GetCurrentTablespaceSegmentLocation(tablespaceId);
+	num_files = CountTablespaceFiles(curr_segment);
+	segmentId = GetCurrentTablespaceSegmentId(curr_segment);
+
+	if (num_files >= limit) /* Check for tablespace overflow */
+	{
+		/* Search pg_tablespace */
+		rel = table_open(TableSpaceRelationId, RowExclusiveLock);
+
+		/* Extract information about original tablespace */
+		ScanKeyInit(&entry[0],
+					Anum_pg_tablespace_oid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(tablespaceId));
+		scan = table_beginscan_catalog(rel, 1, entry);
+		tup = heap_getnext(scan, ForwardScanDirection);
+		if (!HeapTupleIsValid(tup)) /* If tablespace with specified OID was not found, then most likely it was removed */
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("tablespace with OID %d was removed",
+							tablespaceId)));
+		/* Extract tablespace options */
+		relopt = untransformRelOptions(heap_getattr(tup, Anum_pg_tablespace_spcoptions,
+													RelationGetDescr(rel), &isnull));
+		/* Original tablespace name */
+		orig_spcname = pstrdup(NameStr(((Form_pg_tablespace) GETSTRUCT(tup))->spcname));
+		spcowner = ((Form_pg_tablespace) GETSTRUCT(tup))->spcowner;
+
+		table_endscan(scan);
+
+		/* Choose new unused tablespace name by concatanating segno suffix */
+		do
+		{
+			new_spcname = psprintf("%s.%d", orig_spcname, segmentId++);
+			ScanKeyInit(&entry[0],
+						Anum_pg_tablespace_spcname,
+						BTEqualStrategyNumber, F_NAMEEQ,
+						CStringGetDatum(new_spcname));
+			scan = table_beginscan_catalog(rel, 1, entry);
+			tup = heap_getnext(scan, ForwardScanDirection);
+			segment_is_used = HeapTupleIsValid(tup);
+			table_endscan(scan);
+		} while (segment_is_used);
+
+		table_close(rel, RowExclusiveLock);
+
+		stmt = makeNode(CreateTableSpaceStmt);
+		stmt->tablespacename = new_spcname;
+		stmt->owner = makeNode(RoleSpec);
+		stmt->owner->roletype = ROLESPEC_CSTRING;
+		stmt->owner->rolename = GetUserNameFromId(spcowner, false);
+		stmt->location = curr_segment;
+		stmt->options = relopt;
+
+		/* Create new tablespace */
+		segmentId = CreateTableSpace(stmt);
+		Assert(OidIsValid(segmentId));
+	}
+	return segmentId;
+}

src/backend/utils/cache/spccache.c

@@ -221,3 +221,21 @@ get_tablespace_io_concurrency(Oid spcid)
 	else
 		return spc->opts->effective_io_concurrency;
 }
+
+/*
+ * get_tablespace_max_files
+ *
+ *		This value is not locked by the transaction, so this value may
+ *		be changed while a SELECT that has used these values for planning
+ *		is still executing.
+ */
+int
+get_tablespace_max_files(Oid spcid)
+{
+	TableSpaceCacheEntry *spc = get_tablespace(spcid);
+
+	if (!spc->opts || spc->opts->max_files <= 0)
+		return 0; /* no limit by default */
+	else
+		return spc->opts->max_files;
+}

src/include/commands/tablespace.h

@@ -40,6 +40,7 @@ typedef struct TableSpaceOpts
 	float8		random_page_cost;
 	float8		seq_page_cost;
 	int			effective_io_concurrency;
+	int         max_files;
 } TableSpaceOpts;
 
 extern Oid	CreateTableSpace(CreateTableSpaceStmt *stmt);
@@ -53,6 +54,8 @@ extern Oid	GetDefaultTablespace(char relpersistence, bool partitioned);
 
 extern void PrepareTempTablespaces(void);
 
+extern Oid  GetCurrentTablespaceSegment(Oid tablespaceId);
+
 extern Oid	get_tablespace_oid(const char *tablespacename, bool missing_ok);
 extern char *get_tablespace_name(Oid spc_oid);
 

src/include/utils/spccache.h

@@ -16,5 +16,6 @@
 void		get_tablespace_page_costs(Oid spcid, float8 *spc_random_page_cost,
 									  float8 *spc_seq_page_cost);
 int			get_tablespace_io_concurrency(Oid spcid);
+int			get_tablespace_max_files(Oid spcid);
 
 #endif							/* SPCCACHE_H */