CFS branch for PGPROEE9_6

Author: knizhnik

doc/src/sgml/filelist.sgml

@@ -49,6 +49,7 @@
 <!ENTITY config        SYSTEM "config.sgml">
 <!ENTITY user-manag    SYSTEM "user-manag.sgml">
 <!ENTITY wal           SYSTEM "wal.sgml">
+<!ENTITY cfs           SYSTEM "cfs.sgml">
 
 <!-- programmer's guide -->
 <!ENTITY bgworker   SYSTEM "bgworker.sgml">

doc/src/sgml/postgres.sgml

@@ -160,6 +160,7 @@
   &diskusage;
   &wal;
   &regress;
+  &cfs;
 
  </part>
 

doc/src/sgml/ref/create_tablespace.sgml

@@ -105,15 +105,16 @@ CREATE TABLESPACE <replaceable class="parameter">tablespace_name</replaceable>
        <para>
         A tablespace parameter to be set or reset.  Currently, the only
         available parameters are <varname>seq_page_cost</>,
-        <varname>random_page_cost</> and <varname>effective_io_concurrency</>.
-        Setting either value for a particular tablespace will override the
+        <varname>random_page_cost</>, <varname>effective_io_concurrency</> and <varname>compression</>.
+        Setting cost value for a particular tablespace will override the
         planner's usual estimate of the cost of reading pages from tables in
         that tablespace, as established by the configuration parameters of the
         same name (see <xref linkend="guc-seq-page-cost">,
         <xref linkend="guc-random-page-cost">,
         <xref linkend="guc-effective-io-concurrency">).  This may be useful if
         one tablespace is located on a disk which is faster or slower than the
         remainder of the I/O subsystem.
+        Compression is discussed in section <xref linkend="cfs">.
        </para>
       </listitem>
      </varlistentry>

src/backend/Makefile

@@ -43,12 +43,18 @@ OBJS = $(SUBDIROBJS) $(LOCALOBJS) $(top_builddir)/src/port/libpgport_srv.a \
 LIBS := $(filter-out -lpgport -lpgcommon, $(LIBS)) $(LDAP_LIBS_BE)
 
 # The backend doesn't need everything that's in LIBS, however
-LIBS := $(filter-out -lz -lreadline -ledit -ltermcap -lncurses -lcurses, $(LIBS))
+LIBS := $(filter-out -lreadline -ledit -ltermcap -lncurses -lcurses, $(LIBS)) 
 
 ifeq ($(with_systemd),yes)
 LIBS += -lsystemd
 endif
 
+LIBS += -lzstd
+
+
+# Extra compression libraries  
+#LIBS += -llz4 -lsnappy -llzfse
+
 ##########################################################################
 
 all: submake-libpgport submake-schemapg postgres $(POSTGRES_IMP)

src/backend/access/common/reloptions.c

@@ -54,6 +54,15 @@
 
 static relopt_bool boolRelOpts[] =
 {
+	{
+		{
+			"compression",
+			"Enables compression for this table space",
+			RELOPT_KIND_TABLESPACE,
+			AccessExclusiveLock
+		},
+		false
+	},
 	{
 		{
 			"autovacuum_enabled",
@@ -1451,7 +1460,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)},
+		{"compression", RELOPT_TYPE_BOOL, offsetof(TableSpaceOpts, compression)}
 	};
 
 	options = parseRelOptions(reloptions, validate, RELOPT_KIND_TABLESPACE,

src/backend/access/rmgrdesc/tblspcdesc.c

@@ -27,7 +27,7 @@ tblspc_desc(StringInfo buf, XLogReaderState *record)
 	{
 		xl_tblspc_create_rec *xlrec = (xl_tblspc_create_rec *) rec;
 
-		appendStringInfo(buf, "%u \"%s\"", xlrec->ts_id, xlrec->ts_path);
+		appendStringInfo(buf, "%u %s\"%s\"", xlrec->ts_id, xlrec->ts_compressed ? "compressed " : "", xlrec->ts_path);
 	}
 	else if (info == XLOG_TBLSPC_DROP)
 	{

src/backend/commands/dbcommands.c

@@ -61,6 +61,7 @@
 #include "utils/pg_locale.h"
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
+#include "utils/spccache.h"
 #include "utils/tqual.h"
 
 
@@ -1049,6 +1050,8 @@ movedb(const char *dbname, const char *tblspcname)
 	char	   *src_dbpath;
 	char	   *dst_dbpath;
 	DIR		   *dstdir;
+	bool        src_compressed;
+	bool        dst_compressed;
 	struct dirent *xlde;
 	movedb_failure_params fparms;
 
@@ -1122,6 +1125,9 @@ movedb(const char *dbname, const char *tblspcname)
 									 AccessExclusiveLock);
 		return;
 	}
+	
+	src_compressed = is_tablespace_compressed(src_tblspcoid);
+	dst_compressed = is_tablespace_compressed(dst_tblspcoid);
 
 	/*
 	 * Check for other backends in the target database.  (Because we hold the
@@ -1221,7 +1227,11 @@ movedb(const char *dbname, const char *tblspcname)
 		/*
 		 * Copy files from the old tablespace to the new one
 		 */
-		copydir(src_dbpath, dst_dbpath, false);
+		if (src_compressed ^ dst_compressed) { 
+			copyzipdir(src_dbpath, src_compressed, dst_dbpath, dst_compressed);
+		} else { 
+			copydir(src_dbpath, dst_dbpath, false);
+		}
 
 		/*
 		 * Record the filesystem change in XLOG

src/backend/commands/tablespace.c

@@ -48,6 +48,7 @@
 
 #include <unistd.h>
 #include <dirent.h>
+#include <stdio.h>
 #include <sys/types.h>
 #include <sys/stat.h>
 
@@ -65,20 +66,23 @@
 #include "catalog/objectaccess.h"
 #include "catalog/pg_namespace.h"
 #include "catalog/pg_tablespace.h"
+#include "commands/defrem.h"
 #include "commands/comment.h"
 #include "commands/seclabel.h"
 #include "commands/tablecmds.h"
 #include "commands/tablespace.h"
 #include "miscadmin.h"
 #include "postmaster/bgwriter.h"
 #include "storage/fd.h"
+#include "storage/cfs.h"
 #include "storage/lmgr.h"
 #include "storage/standby.h"
 #include "utils/acl.h"
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
 #include "utils/guc.h"
 #include "utils/lsyscache.h"
+#include "utils/spccache.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
 #include "utils/tqual.h"
@@ -90,7 +94,7 @@ char	   *temp_tablespaces = NULL;
 
 
 static void create_tablespace_directories(const char *location,
-							  const Oid tablespaceoid);
+										  const Oid tablespaceoid, bool compressed);
 static bool destroy_tablespace_directories(Oid tablespaceoid, bool redo);
 
 
@@ -222,6 +226,23 @@ TablespaceCreateDbspace(Oid spcNode, Oid dbNode, bool isRedo)
 	pfree(dir);
 }
 
+static bool
+getBoolOption(List* options, char const* name, bool defaultValue)
+{
+	ListCell   *cell;
+	foreach(cell, options)
+	{
+		DefElem    *def = (DefElem *) lfirst(cell);
+		if (strcmp(def->defname, name) == 0) 
+		{
+			return defGetBoolean(def);
+		}
+	}
+	return defaultValue;
+}
+
+
+
 /*
  * Create a table space
  *
@@ -241,6 +262,7 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 	char	   *location;
 	Oid			ownerId;
 	Datum		newOptions;
+	bool        compressed;
 
 	/* Must be super user */
 	if (!superuser())
@@ -256,6 +278,9 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 	else
 		ownerId = GetUserId();
 
+	/* If tablespace should be compressed */
+	compressed = getBoolOption(stmt->options, "compression", false);
+
 	/* Unix-ify the offered path, and strip any trailing slashes */
 	location = pstrdup(stmt->location);
 	canonicalize_path(location);
@@ -355,14 +380,14 @@ CreateTableSpace(CreateTableSpaceStmt *stmt)
 	/* Post creation hook for new tablespace */
 	InvokeObjectPostCreateHook(TableSpaceRelationId, tablespaceoid, 0);
 
-	create_tablespace_directories(location, tablespaceoid);
+	create_tablespace_directories(location, tablespaceoid, getBoolOption(stmt->options, "compression", false));
 
 	/* Record the filesystem change in XLOG */
 	{
 		xl_tblspc_create_rec xlrec;
 
 		xlrec.ts_id = tablespaceoid;
-
+		xlrec.ts_compressed = compressed;
 		XLogBeginInsert();
 		XLogRegisterData((char *) &xlrec,
 						 offsetof(xl_tblspc_create_rec, ts_path));
@@ -562,7 +587,7 @@ DropTableSpace(DropTableSpaceStmt *stmt)
  *	to the specified directory
  */
 static void
-create_tablespace_directories(const char *location, const Oid tablespaceoid)
+create_tablespace_directories(const char *location, const Oid tablespaceoid, bool compressed)
 {
 	char	   *linkloc;
 	char	   *location_with_version_dir;
@@ -626,6 +651,16 @@ create_tablespace_directories(const char *location, const Oid tablespaceoid)
 							location_with_version_dir)));
 	}
 
+	if (compressed)
+	{
+		char* compressionFilePath = psprintf("%s/pg_compression", location_with_version_dir);
+		FILE* comp = fopen(compressionFilePath, "w");
+		elog(LOG, "Create compressed tablespace at %s", location);
+		fputs(cfs_algorithm(), comp);
+		fclose(comp);
+		pfree(compressionFilePath);
+	}
+
 	/*
 	 * In recovery, remove old symlink, in case it points to the wrong place.
 	 */
@@ -664,6 +699,7 @@ destroy_tablespace_directories(Oid tablespaceoid, bool redo)
 {
 	char	   *linkloc;
 	char	   *linkloc_with_version_dir;
+	char       *compression_file;
 	DIR		   *dirdesc;
 	struct dirent *de;
 	char	   *subfile;
@@ -672,6 +708,9 @@ destroy_tablespace_directories(Oid tablespaceoid, bool redo)
 	linkloc_with_version_dir = psprintf("pg_tblspc/%u/%s", tablespaceoid,
 										TABLESPACE_VERSION_DIRECTORY);
 
+	compression_file = psprintf("%s/pg_compression", linkloc_with_version_dir);
+	unlink(compression_file);
+
 	/*
 	 * Check if the tablespace still contains any files.  We try to rmdir each
 	 * per-database directory we find in it.  rmdir failure implies there are
@@ -840,7 +879,8 @@ directory_is_empty(const char *path)
 	while ((de = ReadDir(dirdesc, path)) != NULL)
 	{
 		if (strcmp(de->d_name, ".") == 0 ||
-			strcmp(de->d_name, "..") == 0)
+			strcmp(de->d_name, "..") == 0 ||
+			strcmp(de->d_name, "pg_compression") == 0)
 			continue;
 		FreeDir(dirdesc);
 		return false;
@@ -1000,6 +1040,7 @@ AlterTableSpaceOptions(AlterTableSpaceOptionsStmt *stmt)
 	bool		repl_null[Natts_pg_tablespace];
 	bool		repl_repl[Natts_pg_tablespace];
 	HeapTuple	newtuple;
+	bool        compressed;
 
 	/* Search pg_tablespace */
 	rel = heap_open(TableSpaceRelationId, RowExclusiveLock);
@@ -1018,6 +1059,27 @@ AlterTableSpaceOptions(AlterTableSpaceOptionsStmt *stmt)
 
 	tablespaceoid = HeapTupleGetOid(tup);
 
+	/* If tablespace should be compressed */
+	compressed = getBoolOption(stmt->options, "compression", false);
+	
+	if (compressed ^ is_tablespace_compressed(tablespaceoid)) {
+		char* tbsdir = psprintf("pg_tblspc/%u/%s", tablespaceoid, TABLESPACE_VERSION_DIRECTORY);
+		if (!directory_is_empty(tbsdir)) {
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("It is not possible to toggle compression option for tablespace")));
+		} else { 
+			char* compressionFilePath = psprintf("%s/pg_compression", tbsdir);
+			if (compressed) { 
+				FILE* comp = fopen(compressionFilePath, "w");
+				fputs(cfs_algorithm(), comp);
+				fclose(comp);
+			} else {
+				unlink(compressionFilePath);
+			}
+		}
+	}
+
 	/* Must be owner of the existing object */
 	if (!pg_tablespace_ownercheck(HeapTupleGetOid(tup), GetUserId()))
 		aclcheck_error(ACLCHECK_NOT_OWNER, ACL_KIND_TABLESPACE,
@@ -1478,7 +1540,7 @@ tblspc_redo(XLogReaderState *record)
 		xl_tblspc_create_rec *xlrec = (xl_tblspc_create_rec *) XLogRecGetData(record);
 		char	   *location = xlrec->ts_path;
 
-		create_tablespace_directories(location, xlrec->ts_id);
+		create_tablespace_directories(location, xlrec->ts_id, xlrec->ts_compressed);
 	}
 	else if (info == XLOG_TBLSPC_DROP)
 	{

src/backend/postmaster/postmaster.c

@@ -115,6 +115,7 @@
 #include "postmaster/syslogger.h"
 #include "replication/walsender.h"
 #include "storage/fd.h"
+#include "storage/cfs.h"
 #include "storage/ipc.h"
 #include "storage/pg_shmem.h"
 #include "storage/pmsignal.h"
@@ -502,6 +503,7 @@ typedef struct
 	bool		IsBinaryUpgrade;
 	int			max_safe_fds;
 	int			MaxBackends;
+	CfsState	*cfs_state;
 #ifdef WIN32
 	HANDLE		PostmasterHandle;
 	HANDLE		initial_signal_pipe;
@@ -1238,6 +1240,11 @@ PostmasterMain(int argc, char *argv[])
 	 */
 	autovac_init();
 
+	/*
+	 * Initialize compressed file sysystem support
+	 */
+	cfs_initialize();
+
 	/*
 	 * Load configuration files for client authentication.
 	 */
@@ -5836,6 +5843,8 @@ save_backend_variables(BackendParameters *param, Port *port,
 
 	param->MaxBackends = MaxBackends;
 
+	param->cfs_state = cfs_state;
+
 #ifdef WIN32
 	param->PostmasterHandle = PostmasterHandle;
 	if (!write_duplicated_handle(&param->initial_signal_pipe,
@@ -6068,6 +6077,8 @@ restore_backend_variables(BackendParameters *param, Port *port)
 
 	MaxBackends = param->MaxBackends;
 
+	cfs_state = param->cfs_state;
+
 #ifdef WIN32
 	PostmasterHandle = param->PostmasterHandle;
 	pgwin32_initial_signal_pipe = param->initial_signal_pipe;

src/backend/storage/file/Makefile

@@ -12,6 +12,6 @@ subdir = src/backend/storage/file
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = fd.o buffile.o copydir.o reinit.o
+OBJS = fd.o cfs.o buffile.o copydir.o reinit.o
 
 include $(top_srcdir)/src/backend/common.mk