agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Ildus Kurbangaliev <[email protected]>
Subject: [PATCH 2/8] Add compression catalog tables and the basic infrastructure
Date: Mon, 18 Jun 2018 15:26:13 +0300
---
src/backend/access/Makefile | 4 +-
src/backend/access/common/indextuple.c | 25 +-
src/backend/access/common/reloptions.c | 89 +++
src/backend/access/compression/Makefile | 17 +
src/backend/access/compression/cmapi.c | 96 +++
src/backend/access/heap/heapam.c | 7 +-
src/backend/access/heap/rewriteheap.c | 3 +-
src/backend/access/heap/tuptoaster.c | 475 +++++++++++--
src/backend/access/index/amapi.c | 47 +-
src/backend/bootstrap/bootstrap.c | 1 +
src/backend/catalog/Makefile | 2 +-
src/backend/catalog/dependency.c | 11 +-
src/backend/catalog/heap.c | 3 +
src/backend/catalog/index.c | 4 +-
src/backend/catalog/objectaddress.c | 57 ++
src/backend/commands/Makefile | 4 +-
src/backend/commands/alter.c | 1 +
src/backend/commands/amcmds.c | 84 +++
src/backend/commands/compressioncmds.c | 653 ++++++++++++++++++
src/backend/commands/event_trigger.c | 1 +
src/backend/commands/foreigncmds.c | 46 +-
src/backend/commands/tablecmds.c | 336 ++++++++-
src/backend/nodes/copyfuncs.c | 16 +
src/backend/nodes/equalfuncs.c | 14 +
src/backend/nodes/nodeFuncs.c | 10 +
src/backend/nodes/outfuncs.c | 14 +
src/backend/parser/parse_utilcmd.c | 7 +
.../replication/logical/reorderbuffer.c | 2 +-
src/backend/utils/adt/pseudotypes.c | 1 +
src/backend/utils/cache/syscache.c | 12 +
src/include/access/cmapi.h | 80 +++
src/include/access/reloptions.h | 3 +-
src/include/access/tuptoaster.h | 42 +-
src/include/catalog/dependency.h | 5 +-
src/include/catalog/indexing.h | 5 +
src/include/catalog/pg_attr_compression.dat | 23 +
src/include/catalog/pg_attr_compression.h | 53 ++
src/include/catalog/pg_attribute.h | 7 +-
src/include/catalog/pg_class.dat | 2 +-
src/include/catalog/pg_proc.dat | 10 +
src/include/catalog/pg_type.dat | 5 +
src/include/commands/defrem.h | 14 +
src/include/nodes/nodes.h | 3 +-
src/include/postgres.h | 26 +-
src/include/utils/syscache.h | 1 +
src/tools/pgindent/typedefs.list | 3 +
46 files changed, 2129 insertions(+), 195 deletions(-)
create mode 100644 src/backend/access/compression/Makefile
create mode 100644 src/backend/access/compression/cmapi.c
create mode 100644 src/backend/commands/compressioncmds.c
create mode 100644 src/include/access/cmapi.h
create mode 100644 src/include/catalog/pg_attr_compression.dat
create mode 100644 src/include/catalog/pg_attr_compression.h
diff --git a/src/backend/access/Makefile b/src/backend/access/Makefile
index bd93a6a8d1..4cd4fd9715 100644
--- a/src/backend/access/Makefile
+++ b/src/backend/access/Makefile
@@ -8,7 +8,7 @@ subdir = src/backend/access
top_builddir = ../../..
include $(top_builddir)/src/Makefile.global
-SUBDIRS = brin common gin gist hash heap index nbtree rmgrdesc spgist \
- tablesample transam
+SUBDIRS = brin common compression gin gist hash heap index nbtree \
+ rmgrdesc spgist tablesample transam
include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c
index aa52a96259..9a1cebbe2f 100644
--- a/src/backend/access/common/indextuple.c
+++ b/src/backend/access/common/indextuple.c
@@ -77,13 +77,30 @@ index_form_tuple(TupleDesc tupleDescriptor,
/*
* If value is stored EXTERNAL, must fetch it so we are not depending
- * on outside storage. This should be improved someday.
+ * on outside storage. This should be improved someday. If value also
+ * was compressed by custom compression method then we should
+ * decompress it too.
*/
if (VARATT_IS_EXTERNAL(DatumGetPointer(values[i])))
+ {
+ struct varatt_external toast_pointer;
+
+ VARATT_EXTERNAL_GET_POINTER(toast_pointer, DatumGetPointer(values[i]));
+ if (VARATT_EXTERNAL_IS_CUSTOM_COMPRESSED(toast_pointer))
+ untoasted_values[i] =
+ PointerGetDatum(heap_tuple_untoast_attr((struct varlena *)
+ DatumGetPointer(values[i])));
+ else
+ untoasted_values[i] =
+ PointerGetDatum(heap_tuple_fetch_attr((struct varlena *)
+ DatumGetPointer(values[i])));
+ untoasted_free[i] = true;
+ }
+ else if (VARATT_IS_CUSTOM_COMPRESSED(DatumGetPointer(values[i])))
{
untoasted_values[i] =
- PointerGetDatum(heap_tuple_fetch_attr((struct varlena *)
- DatumGetPointer(values[i])));
+ PointerGetDatum(heap_tuple_untoast_attr((struct varlena *)
+ DatumGetPointer(values[i])));
untoasted_free[i] = true;
}
@@ -95,7 +112,7 @@ index_form_tuple(TupleDesc tupleDescriptor,
VARSIZE(DatumGetPointer(untoasted_values[i])) > TOAST_INDEX_TARGET &&
(att->attstorage == 'x' || att->attstorage == 'm'))
{
- Datum cvalue = toast_compress_datum(untoasted_values[i]);
+ Datum cvalue = toast_compress_datum(untoasted_values[i], InvalidOid);
if (DatumGetPointer(cvalue) != NULL)
{
diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c
index db84da0678..bf836305a9 100644
--- a/src/backend/access/common/reloptions.c
+++ b/src/backend/access/common/reloptions.c
@@ -962,11 +962,100 @@ untransformRelOptions(Datum options)
val = (Node *) makeString(pstrdup(p));
}
result = lappend(result, makeDefElem(pstrdup(s), val, -1));
+ pfree(s);
}
return result;
}
+/*
+ * helper function for qsort to compare DefElem
+ */
+static int
+compare_options(const void *a, const void *b)
+{
+ DefElem *da = (DefElem *) lfirst(*(ListCell **) a);
+ DefElem *db = (DefElem *) lfirst(*(ListCell **) b);
+
+ return strcmp(da->defname, db->defname);
+}
+
+/*
+ * Convert a DefElem list to the text array format that is used in
+ * pg_foreign_data_wrapper, pg_foreign_server, pg_user_mapping,
+ * pg_foreign_table and pg_attr_compression
+ *
+ * Returns the array in the form of a Datum, or PointerGetDatum(NULL)
+ * if the list is empty.
+ *
+ * Note: The array is usually stored to database without further
+ * processing, hence any validation should be done before this
+ * conversion.
+ */
+Datum
+optionListToArray(List *options, bool sorted)
+{
+ ArrayBuildState *astate = NULL;
+ ListCell *cell;
+ List *resoptions = NIL;
+ int len = list_length(options);
+
+ /* sort by option name if needed */
+ if (sorted && len > 1)
+ resoptions = list_qsort(options, compare_options);
+ else
+ resoptions = options;
+
+ foreach(cell, resoptions)
+ {
+ DefElem *def = lfirst(cell);
+ const char *value;
+ Size len;
+ text *t;
+
+ value = defGetString(def);
+ len = VARHDRSZ + strlen(def->defname) + 1 + strlen(value);
+ t = palloc(len + 1);
+ SET_VARSIZE(t, len);
+ sprintf(VARDATA(t), "%s=%s", def->defname, value);
+
+ astate = accumArrayResult(astate, PointerGetDatum(t),
+ false, TEXTOID,
+ CurrentMemoryContext);
+ }
+
+ /* free if the the modified version of list was used */
+ if (resoptions != options)
+ list_free(resoptions);
+
+ if (astate)
+ return makeArrayResult(astate, CurrentMemoryContext);
+
+ return PointerGetDatum(NULL);
+}
+
+/*
+ * Return human readable list of reloptions
+ */
+char *
+formatRelOptions(List *options)
+{
+ StringInfoData buf;
+ ListCell *cell;
+
+ initStringInfo(&buf);
+
+ foreach(cell, options)
+ {
+ DefElem *def = (DefElem *) lfirst(cell);
+
+ appendStringInfo(&buf, "%s%s=%s", buf.len > 0 ? ", " : "",
+ def->defname, defGetString(def));
+ }
+
+ return buf.data;
+}
+
/*
* Extract and parse reloptions from a pg_class tuple.
*
diff --git a/src/backend/access/compression/Makefile b/src/backend/access/compression/Makefile
new file mode 100644
index 0000000000..a09dc787ed
--- /dev/null
+++ b/src/backend/access/compression/Makefile
@@ -0,0 +1,17 @@
+#-------------------------------------------------------------------------
+#
+# Makefile--
+# Makefile for access/compression
+#
+# IDENTIFICATION
+# src/backend/access/compression/Makefile
+#
+#-------------------------------------------------------------------------
+
+subdir = src/backend/access/compression
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+
+OBJS = cmapi.o
+
+include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/access/compression/cmapi.c b/src/backend/access/compression/cmapi.c
new file mode 100644
index 0000000000..504da0638f
--- /dev/null
+++ b/src/backend/access/compression/cmapi.c
@@ -0,0 +1,96 @@
+/*-------------------------------------------------------------------------
+ *
+ * compression/cmapi.c
+ * Functions for compression access methods
+ *
+ * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ * src/backend/access/compression/cmapi.c
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+#include "fmgr.h"
+#include "access/cmapi.h"
+#include "access/htup_details.h"
+#include "access/reloptions.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_attr_compression.h"
+#include "commands/defrem.h"
+#include "utils/syscache.h"
+
+/*
+ * InvokeCompressionAmHandler - call the specified access method handler routine to get
+ * its CompressionAmRoutine struct, which will be palloc'd in the caller's context.
+ */
+CompressionAmRoutine *
+InvokeCompressionAmHandler(Oid amhandler)
+{
+ Datum datum;
+ CompressionAmRoutine *routine;
+
+ datum = OidFunctionCall0(amhandler);
+ routine = (CompressionAmRoutine *) DatumGetPointer(datum);
+
+ if (routine == NULL || !IsA(routine, CompressionAmRoutine))
+ elog(ERROR, "compression method handler function %u "
+ "did not return an CompressionAmRoutine struct",
+ amhandler);
+
+ return routine;
+}
+
+/*
+ * GetAttrCompressionOptions
+ *
+ * Parse array of attribute compression options and return it as a list.
+ */
+List *
+GetAttrCompressionOptions(Oid acoid)
+{
+ HeapTuple tuple;
+ List *result = NIL;
+ bool isnull;
+ Datum acoptions;
+
+ tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for an attribute compression %u", acoid);
+
+ /* options could be NULL, so we can't use form struct */
+ acoptions = SysCacheGetAttr(ATTCOMPRESSIONOID, tuple,
+ Anum_pg_attr_compression_acoptions, &isnull);
+
+ if (!isnull)
+ result = untransformRelOptions(acoptions);
+
+ ReleaseSysCache(tuple);
+ return result;
+}
+
+/*
+ * GetAttrCompressionAmOid
+ *
+ * Return access method Oid by attribute compression Oid
+ */
+Oid
+GetAttrCompressionAmOid(Oid acoid)
+{
+ Oid result;
+ HeapTuple tuple;
+ Form_pg_attr_compression acform;
+
+ /* extract access method Oid */
+ tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+ acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+ result = get_compression_am_oid(NameStr(acform->acname), false);
+ ReleaseSysCache(tuple);
+
+ return result;
+}
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 72395a50b8..fdac4bbb67 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2684,8 +2684,9 @@ heap_prepare_insert(Relation relation, HeapTuple tup, TransactionId xid,
Assert(!HeapTupleHasExternal(tup));
return tup;
}
- else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
- return toast_insert_or_update(relation, tup, NULL, options);
+ else if (HeapTupleHasExternal(tup)
+ || tup->t_len > TOAST_TUPLE_THRESHOLD)
+ return toast_insert_or_update(relation, tup, NULL, options, NULL);
else
return tup;
}
@@ -4129,7 +4130,7 @@ l2:
if (need_toast)
{
/* Note we always use WAL and FSM during updates */
- heaptup = toast_insert_or_update(relation, newtup, &oldtup, 0);
+ heaptup = toast_insert_or_update(relation, newtup, &oldtup, 0, NULL);
newtupsize = MAXALIGN(heaptup->t_len);
}
else
diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index ed7ba181c7..802497a4d9 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -655,7 +655,8 @@ raw_heap_insert(RewriteState state, HeapTuple tup)
heaptup = toast_insert_or_update(state->rs_new_rel, tup, NULL,
HEAP_INSERT_SKIP_FSM |
(state->rs_use_wal ?
- 0 : HEAP_INSERT_SKIP_WAL));
+ 0 : HEAP_INSERT_SKIP_WAL),
+ NULL);
else
heaptup = tup;
diff --git a/src/backend/access/heap/tuptoaster.c b/src/backend/access/heap/tuptoaster.c
index cd42c50b09..413b0b2b9c 100644
--- a/src/backend/access/heap/tuptoaster.c
+++ b/src/backend/access/heap/tuptoaster.c
@@ -30,17 +30,25 @@
#include <unistd.h>
#include <fcntl.h>
+#include "access/cmapi.h"
#include "access/genam.h"
#include "access/heapam.h"
+#include "access/reloptions.h"
#include "access/tuptoaster.h"
#include "access/xact.h"
#include "catalog/catalog.h"
+#include "catalog/pg_am.h"
+#include "commands/defrem.h"
#include "common/pg_lzcompress.h"
#include "miscadmin.h"
#include "utils/expandeddatum.h"
#include "utils/fmgroids.h"
+#include "utils/hsearch.h"
+#include "utils/inval.h"
+#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
+#include "utils/syscache.h"
#include "utils/typcache.h"
#include "utils/tqual.h"
@@ -53,19 +61,51 @@
typedef struct toast_compress_header
{
int32 vl_len_; /* varlena header (do not touch directly!) */
- int32 rawsize;
+ uint32 info; /* flags (2 bits) and rawsize */
} toast_compress_header;
+/*
+ * If the compression method were used, then data also contains
+ * Oid of compression options
+ */
+typedef struct toast_compress_header_custom
+{
+ int32 vl_len_; /* varlena header (do not touch directly!) */
+ uint32 info; /* flags (2 high bits) and rawsize */
+ Oid cmid; /* Oid from pg_attr_compression */
+} toast_compress_header_custom;
+
+static HTAB *amoptions_cache = NULL;
+static MemoryContext amoptions_cache_mcxt = NULL;
+
+#define RAWSIZEMASK (0x3FFFFFFFU)
+
/*
* Utilities for manipulation of header information for compressed
* toast entries.
+ *
+ * Since version 11 TOAST_COMPRESS_SET_RAWSIZE also marks compressed
+ * varlenas as custom compressed. Such varlenas will contain 0x02 (0b10) in
+ * two highest bits.
*/
-#define TOAST_COMPRESS_HDRSZ ((int32) sizeof(toast_compress_header))
-#define TOAST_COMPRESS_RAWSIZE(ptr) (((toast_compress_header *) (ptr))->rawsize)
+#define TOAST_COMPRESS_HDRSZ ((int32) sizeof(toast_compress_header))
+#define TOAST_COMPRESS_HDRSZ_CUSTOM ((int32) sizeof(toast_compress_header_custom))
+#define TOAST_COMPRESS_RAWSIZE(ptr) (((toast_compress_header *) (ptr))->info & RAWSIZEMASK)
#define TOAST_COMPRESS_RAWDATA(ptr) \
(((char *) (ptr)) + TOAST_COMPRESS_HDRSZ)
#define TOAST_COMPRESS_SET_RAWSIZE(ptr, len) \
- (((toast_compress_header *) (ptr))->rawsize = (len))
+do { \
+ Assert(len > 0 && len <= RAWSIZEMASK); \
+ ((toast_compress_header *) (ptr))->info = (len) | (0x02 << 30); \
+} while (0)
+#define TOAST_COMPRESS_SET_CMID(ptr, oid) \
+ (((toast_compress_header_custom *) (ptr))->cmid = (oid))
+
+#define VARATT_EXTERNAL_SET_CUSTOM(toast_pointer) \
+do { \
+ ((toast_pointer).va_extinfo |= (1 << 31)); \
+ ((toast_pointer).va_extinfo &= ~(1 << 30)); \
+} while (0)
static void toast_delete_datum(Relation rel, Datum value, bool is_speculative);
static Datum toast_save_datum(Relation rel, Datum value,
@@ -83,7 +123,8 @@ static int toast_open_indexes(Relation toastrel,
static void toast_close_indexes(Relation *toastidxs, int num_indexes,
LOCKMODE lock);
static void init_toast_snapshot(Snapshot toast_snapshot);
-
+static void init_amoptions_cache(void);
+static bool attr_compression_options_are_equal(Oid acoid1, Oid acoid2);
/* ----------
* heap_tuple_fetch_attr -
@@ -421,7 +462,7 @@ toast_datum_size(Datum value)
struct varatt_external toast_pointer;
VARATT_EXTERNAL_GET_POINTER(toast_pointer, attr);
- result = toast_pointer.va_extsize;
+ result = VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer);
}
else if (VARATT_IS_EXTERNAL_INDIRECT(attr))
{
@@ -511,6 +552,98 @@ toast_delete(Relation rel, HeapTuple oldtup, bool is_speculative)
}
}
+/* ----------
+ * toast_prepare_varlena
+ *
+ * Untoast or fetch varlena if needed. This function could return
+ * compressed varlena.
+ */
+static struct varlena *
+toast_prepare_varlena(Form_pg_attribute att, struct varlena *value,
+ List *preserved_amoids, bool *modified)
+{
+ struct varlena *tmpval = NULL;
+
+ *modified = false;
+
+ /*
+ * We took care of UPDATE, so any external value we find still in the
+ * tuple must be someone else's that we cannot reuse (this includes the
+ * case of an out-of-line in-memory datum). Fetch it back (without
+ * decompression, unless we are forcing PLAIN storage or it has a custom
+ * compression). If necessary, we'll push it out as a new external value
+ * below.
+ */
+ if (VARATT_IS_EXTERNAL(value))
+ {
+ *modified = true;
+ if (att->attstorage == 'p')
+ {
+ value = heap_tuple_untoast_attr(value);
+
+ /* untoasted value does not need any further work */
+ return value;
+ }
+ else
+ value = heap_tuple_fetch_attr(value);
+ }
+
+ /*
+ * Process custom compressed datum.
+ *
+ * 1) If destination column has identical compression move the data as is
+ * and only change attribute compression Oid. 2) If it's rewrite from
+ * ALTER command check list of preserved compression access methods. 3) In
+ * other cases just untoast the datum.
+ */
+ if (VARATT_IS_CUSTOM_COMPRESSED(value))
+ {
+ bool storage_ok = (att->attstorage == 'm' || att->attstorage == 'x');
+ toast_compress_header_custom *hdr;
+
+ hdr = (toast_compress_header_custom *) value;
+
+ /* nothing to do */
+ if (storage_ok && hdr->cmid == att->attcompression)
+ return value;
+
+ if (storage_ok &&
+ OidIsValid(att->attcompression) &&
+ attr_compression_options_are_equal(att->attcompression, hdr->cmid))
+ {
+ /* identical compression, just change Oid to new one */
+ tmpval = palloc(VARSIZE(value));
+ memcpy(tmpval, value, VARSIZE(value));
+ TOAST_COMPRESS_SET_CMID(tmpval, att->attcompression);
+ }
+ else if (preserved_amoids != NULL)
+ {
+ Oid amoid = GetAttrCompressionAmOid(hdr->cmid);
+
+ /* decompress the value if it's not in preserved list */
+ if (!list_member_oid(preserved_amoids, amoid))
+ tmpval = heap_tuple_untoast_attr(value);
+ }
+ else
+ /* just decompress the value */
+ tmpval = heap_tuple_untoast_attr(value);
+
+ }
+ else if (OidIsValid(att->attcompression))
+ tmpval = heap_tuple_untoast_attr(value);
+
+ if (tmpval && tmpval != value)
+ {
+ /* if value was external we need to free fetched data */
+ if (*modified)
+ pfree(value);
+
+ value = tmpval;
+ *modified = true;
+ }
+
+ return value;
+}
/* ----------
* toast_insert_or_update -
@@ -522,6 +655,8 @@ toast_delete(Relation rel, HeapTuple oldtup, bool is_speculative)
* newtup: the candidate new tuple to be inserted
* oldtup: the old row version for UPDATE, or NULL for INSERT
* options: options to be passed to heap_insert() for toast rows
+ * preserved_am_info: hash table with attnum as key, for lists of compression
+ * methods which should be preserved on decompression.
* Result:
* either newtup if no toasting is needed, or a palloc'd modified tuple
* that is what should actually get stored
@@ -532,7 +667,7 @@ toast_delete(Relation rel, HeapTuple oldtup, bool is_speculative)
*/
HeapTuple
toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
- int options)
+ int options, HTAB *preserved_am_info)
{
HeapTuple result_tuple;
TupleDesc tupleDesc;
@@ -667,27 +802,37 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
*/
if (att->attlen == -1)
{
+ bool modified = false;
+ List *preserved_amoids = NIL;
+
/*
* If the table's attribute says PLAIN always, force it so.
*/
if (att->attstorage == 'p')
toast_action[i] = 'p';
+ if (VARATT_IS_EXTERNAL(new_value))
+ toast_oldexternal[i] = new_value;
+
/*
- * We took care of UPDATE above, so any external value we find
- * still in the tuple must be someone else's that we cannot reuse
- * (this includes the case of an out-of-line in-memory datum).
- * Fetch it back (without decompression, unless we are forcing
- * PLAIN storage). If necessary, we'll push it out as a new
- * external value below.
+ * If it's ALTER SET COMPRESSION command we should check that
+ * access method of compression in preserved list.
*/
- if (VARATT_IS_EXTERNAL(new_value))
+ if (preserved_am_info != NULL)
+ {
+ bool found;
+ AttrCmPreservedInfo *pinfo;
+
+ pinfo = hash_search(preserved_am_info, &att->attnum,
+ HASH_FIND, &found);
+ if (found)
+ preserved_amoids = pinfo->preserved_amoids;
+ }
+
+ new_value = toast_prepare_varlena(att,
+ new_value, preserved_amoids, &modified);
+ if (modified)
{
- toast_oldexternal[i] = new_value;
- if (att->attstorage == 'p')
- new_value = heap_tuple_untoast_attr(new_value);
- else
- new_value = heap_tuple_fetch_attr(new_value);
toast_values[i] = PointerGetDatum(new_value);
toast_free[i] = true;
need_change = true;
@@ -741,12 +886,14 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
Datum old_value;
Datum new_value;
+ Form_pg_attribute att;
+
/*
* Search for the biggest yet unprocessed internal attribute
*/
for (i = 0; i < numAttrs; i++)
{
- Form_pg_attribute att = TupleDescAttr(tupleDesc, i);
+ char attstorage;
if (toast_action[i] != ' ')
continue;
@@ -754,7 +901,9 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
continue; /* can't happen, toast_action would be 'p' */
if (VARATT_IS_COMPRESSED(DatumGetPointer(toast_values[i])))
continue;
- if (att->attstorage != 'x' && att->attstorage != 'e')
+
+ attstorage = (TupleDescAttr(tupleDesc, i))->attstorage;
+ if (attstorage != 'x' && attstorage != 'e')
continue;
if (toast_sizes[i] > biggest_size)
{
@@ -770,10 +919,11 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
* Attempt to compress it inline, if it has attstorage 'x'
*/
i = biggest_attno;
- if (TupleDescAttr(tupleDesc, i)->attstorage == 'x')
+ att = TupleDescAttr(tupleDesc, i);
+ if (att->attstorage == 'x')
{
old_value = toast_values[i];
- new_value = toast_compress_datum(old_value);
+ new_value = toast_compress_datum(old_value, att->attcompression);
if (DatumGetPointer(new_value) != NULL)
{
@@ -914,7 +1064,8 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup,
*/
i = biggest_attno;
old_value = toast_values[i];
- new_value = toast_compress_datum(old_value);
+ new_value = toast_compress_datum(old_value,
+ TupleDescAttr(tupleDesc, i)->attcompression);
if (DatumGetPointer(new_value) != NULL)
{
@@ -1353,6 +1504,18 @@ toast_build_flattened_tuple(TupleDesc tupleDesc,
return new_tuple;
}
+/* ---------
+ * toast_set_compressed_datum_info
+ *
+ * Save metadata in compressed datum. This information will required
+ * for decompression.
+ */
+void
+toast_set_compressed_datum_info(struct varlena *val, Oid cmid, int32 rawsize)
+{
+ TOAST_COMPRESS_SET_RAWSIZE(val, rawsize);
+ TOAST_COMPRESS_SET_CMID(val, cmid);
+}
/* ----------
* toast_compress_datum -
@@ -1368,54 +1531,47 @@ toast_build_flattened_tuple(TupleDesc tupleDesc,
* ----------
*/
Datum
-toast_compress_datum(Datum value)
+toast_compress_datum(Datum value, Oid acoid)
{
- struct varlena *tmp;
- int32 valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value));
- int32 len;
+ struct varlena *tmp = NULL;
+ int32 valsize;
+ CompressionAmOptions *cmoptions = NULL;
Assert(!VARATT_IS_EXTERNAL(DatumGetPointer(value)));
Assert(!VARATT_IS_COMPRESSED(DatumGetPointer(value)));
- /*
- * No point in wasting a palloc cycle if value size is out of the allowed
- * range for compression
- */
- if (valsize < PGLZ_strategy_default->min_input_size ||
- valsize > PGLZ_strategy_default->max_input_size)
- return PointerGetDatum(NULL);
+ /* Fallback to default compression if not specified */
+ if (!OidIsValid(acoid))
+ acoid = DefaultCompressionOid;
- tmp = (struct varlena *) palloc(PGLZ_MAX_OUTPUT(valsize) +
- TOAST_COMPRESS_HDRSZ);
+ cmoptions = lookup_compression_am_options(acoid);
+ tmp = cmoptions->amroutine->cmcompress(cmoptions, (const struct varlena *) value);
+ if (!tmp)
+ return PointerGetDatum(NULL);
/*
- * We recheck the actual size even if pglz_compress() reports success,
- * because it might be satisfied with having saved as little as one byte
- * in the compressed data --- which could turn into a net loss once you
- * consider header and alignment padding. Worst case, the compressed
- * format might require three padding bytes (plus header, which is
- * included in VARSIZE(tmp)), whereas the uncompressed format would take
- * only one header byte and no padding if the value is short enough. So
- * we insist on a savings of more than 2 bytes to ensure we have a gain.
+ * We recheck the actual size even if compression function reports
+ * success, because it might be satisfied with having saved as little as
+ * one byte in the compressed data --- which could turn into a net loss
+ * once you consider header and alignment padding. Worst case, the
+ * compressed format might require three padding bytes (plus header, which
+ * is included in VARSIZE(tmp)), whereas the uncompressed format would
+ * take only one header byte and no padding if the value is short enough.
+ * So we insist on a savings of more than 2 bytes to ensure we have a
+ * gain.
*/
- len = pglz_compress(VARDATA_ANY(DatumGetPointer(value)),
- valsize,
- TOAST_COMPRESS_RAWDATA(tmp),
- PGLZ_strategy_default);
- if (len >= 0 &&
- len + TOAST_COMPRESS_HDRSZ < valsize - 2)
+ valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value));
+
+ if (VARSIZE(tmp) < valsize - 2)
{
- TOAST_COMPRESS_SET_RAWSIZE(tmp, valsize);
- SET_VARSIZE_COMPRESSED(tmp, len + TOAST_COMPRESS_HDRSZ);
/* successful compression */
+ toast_set_compressed_datum_info(tmp, cmoptions->acoid, valsize);
return PointerGetDatum(tmp);
}
- else
- {
- /* incompressible data */
- pfree(tmp);
- return PointerGetDatum(NULL);
- }
+
+ /* incompressible data */
+ pfree(tmp);
+ return PointerGetDatum(NULL);
}
@@ -1510,19 +1666,20 @@ toast_save_datum(Relation rel, Datum value,
&num_indexes);
/*
- * Get the data pointer and length, and compute va_rawsize and va_extsize.
+ * Get the data pointer and length, and compute va_rawsize and va_extinfo.
*
* va_rawsize is the size of the equivalent fully uncompressed datum, so
* we have to adjust for short headers.
*
- * va_extsize is the actual size of the data payload in the toast records.
+ * va_extinfo contains the actual size of the data payload in the toast
+ * records.
*/
if (VARATT_IS_SHORT(dval))
{
data_p = VARDATA_SHORT(dval);
data_todo = VARSIZE_SHORT(dval) - VARHDRSZ_SHORT;
toast_pointer.va_rawsize = data_todo + VARHDRSZ; /* as if not short */
- toast_pointer.va_extsize = data_todo;
+ toast_pointer.va_extinfo = data_todo; /* no flags */
}
else if (VARATT_IS_COMPRESSED(dval))
{
@@ -1530,7 +1687,10 @@ toast_save_datum(Relation rel, Datum value,
data_todo = VARSIZE(dval) - VARHDRSZ;
/* rawsize in a compressed datum is just the size of the payload */
toast_pointer.va_rawsize = VARRAWSIZE_4B_C(dval) + VARHDRSZ;
- toast_pointer.va_extsize = data_todo;
+ toast_pointer.va_extinfo = data_todo;
+ if (VARATT_IS_CUSTOM_COMPRESSED(dval))
+ VARATT_EXTERNAL_SET_CUSTOM(toast_pointer);
+
/* Assert that the numbers look like it's compressed */
Assert(VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer));
}
@@ -1539,7 +1699,7 @@ toast_save_datum(Relation rel, Datum value,
data_p = VARDATA(dval);
data_todo = VARSIZE(dval) - VARHDRSZ;
toast_pointer.va_rawsize = VARSIZE(dval);
- toast_pointer.va_extsize = data_todo;
+ toast_pointer.va_extinfo = data_todo; /* no flags */
}
/*
@@ -1899,7 +2059,7 @@ toast_fetch_datum(struct varlena *attr)
/* Must copy to access aligned fields */
VARATT_EXTERNAL_GET_POINTER(toast_pointer, attr);
- ressize = toast_pointer.va_extsize;
+ ressize = VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer);
numchunks = ((ressize - 1) / TOAST_MAX_CHUNK_SIZE) + 1;
result = (struct varlena *) palloc(ressize + VARHDRSZ);
@@ -2084,7 +2244,7 @@ toast_fetch_datum_slice(struct varlena *attr, int32 sliceoffset, int32 length)
*/
Assert(!VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer));
- attrsize = toast_pointer.va_extsize;
+ attrsize = VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer);
totalchunks = ((attrsize - 1) / TOAST_MAX_CHUNK_SIZE) + 1;
if (sliceoffset >= attrsize)
@@ -2280,15 +2440,26 @@ toast_decompress_datum(struct varlena *attr)
Assert(VARATT_IS_COMPRESSED(attr));
- result = (struct varlena *)
- palloc(TOAST_COMPRESS_RAWSIZE(attr) + VARHDRSZ);
- SET_VARSIZE(result, TOAST_COMPRESS_RAWSIZE(attr) + VARHDRSZ);
+ if (VARATT_IS_CUSTOM_COMPRESSED(attr))
+ {
+ CompressionAmOptions *cmoptions;
+ toast_compress_header_custom *hdr;
- if (pglz_decompress(TOAST_COMPRESS_RAWDATA(attr),
- VARSIZE(attr) - TOAST_COMPRESS_HDRSZ,
- VARDATA(result),
- TOAST_COMPRESS_RAWSIZE(attr)) < 0)
- elog(ERROR, "compressed data is corrupted");
+ hdr = (toast_compress_header_custom *) attr;
+ cmoptions = lookup_compression_am_options(hdr->cmid);
+ result = cmoptions->amroutine->cmdecompress(cmoptions, attr);
+ }
+ else
+ {
+ result = (struct varlena *)
+ palloc(TOAST_COMPRESS_RAWSIZE(attr) + VARHDRSZ);
+ SET_VARSIZE(result, TOAST_COMPRESS_RAWSIZE(attr) + VARHDRSZ);
+ if (pglz_decompress(TOAST_COMPRESS_RAWDATA(attr),
+ VARSIZE(attr) - TOAST_COMPRESS_HDRSZ,
+ VARDATA(result),
+ TOAST_COMPRESS_RAWSIZE(attr)) < 0)
+ elog(ERROR, "compressed data is corrupted");
+ }
return result;
}
@@ -2390,3 +2561,159 @@ init_toast_snapshot(Snapshot toast_snapshot)
InitToastSnapshot(*toast_snapshot, snapshot->lsn, snapshot->whenTaken);
}
+
+/* ----------
+ * remove_cached_amoptions
+ *
+ * Remove specified compression options from cache
+ */
+static void
+remove_cached_amoptions(CompressionAmOptions *entry)
+{
+ MemoryContextDelete(entry->mcxt);
+
+ if (hash_search(amoptions_cache, (void *) &entry->acoid,
+ HASH_REMOVE, NULL) == NULL)
+ elog(ERROR, "hash table corrupted");
+}
+
+/* ----------
+ * invalidate_amoptions_cache
+ *
+ * Flush cache entries when pg_attr_compression is updated.
+ */
+static void
+invalidate_amoptions_cache(Datum arg, int cacheid, uint32 hashvalue)
+{
+ HASH_SEQ_STATUS status;
+ CompressionAmOptions *entry;
+
+ hash_seq_init(&status, amoptions_cache);
+ while ((entry = (CompressionAmOptions *) hash_seq_search(&status)) != NULL)
+ remove_cached_amoptions(entry);
+}
+
+/* ----------
+ * init_amoptions_cache
+ *
+ * Initialize a local cache for attribute compression options.
+ */
+static void
+init_amoptions_cache(void)
+{
+ HASHCTL ctl;
+
+ amoptions_cache_mcxt = AllocSetContextCreate(TopMemoryContext,
+ "attr compression cache",
+ ALLOCSET_DEFAULT_SIZES);
+ MemSet(&ctl, 0, sizeof(ctl));
+ ctl.keysize = sizeof(Oid);
+ ctl.entrysize = sizeof(CompressionAmOptions);
+ ctl.hcxt = amoptions_cache_mcxt;
+ amoptions_cache = hash_create("attr compression cache", 100, &ctl,
+ HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+
+ /* Set up invalidation callback on pg_attr_compression. */
+ CacheRegisterSyscacheCallback(ATTCOMPRESSIONOID,
+ invalidate_amoptions_cache,
+ (Datum) 0);
+}
+
+/* ----------
+ * lookup_compression_am_options
+ *
+ * Return or generate cache record for attribute compression.
+ */
+CompressionAmOptions *
+lookup_compression_am_options(Oid acoid)
+{
+ bool found,
+ optisnull;
+ CompressionAmOptions *result;
+ Datum acoptions;
+ Form_pg_attr_compression acform;
+ HeapTuple tuple;
+ Oid amoid;
+ regproc amhandler;
+
+ /*
+ * This call could invalidate caches so we need to call it before
+ * we're putting something to cache.
+ */
+ tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+ acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+ amoid = get_compression_am_oid(NameStr(acform->acname), false);
+ amhandler = get_am_handler_oid(amoid, AMTYPE_COMPRESSION, false);
+ acoptions = SysCacheGetAttr(ATTCOMPRESSIONOID, tuple,
+ Anum_pg_attr_compression_acoptions, &optisnull);
+
+ if (!amoptions_cache)
+ init_amoptions_cache();
+
+ result = hash_search(amoptions_cache, &acoid, HASH_ENTER, &found);
+ if (!found)
+ {
+ MemoryContext oldcxt = CurrentMemoryContext;
+
+ result->mcxt = AllocSetContextCreateExtended(amoptions_cache_mcxt,
+ "compression am options",
+ ALLOCSET_DEFAULT_SIZES);
+
+ PG_TRY();
+ {
+ result->acoid = acoid;
+ result->amoid = amoid;
+
+ MemoryContextSwitchTo(result->mcxt);
+ result->amroutine = InvokeCompressionAmHandler(amhandler);
+ if (optisnull)
+ result->acoptions = NIL;
+ else
+ result->acoptions = untransformRelOptions(acoptions);
+ result->acstate = result->amroutine->cminitstate ?
+ result->amroutine->cminitstate(acoid, result->acoptions) : NULL;
+ }
+ PG_CATCH();
+ {
+ MemoryContextSwitchTo(oldcxt);
+ remove_cached_amoptions(result);
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
+ MemoryContextSwitchTo(oldcxt);
+ }
+
+ ReleaseSysCache(tuple);
+
+ if (!result->amroutine)
+ /* should not happen but need to check if something goes wrong */
+ elog(ERROR, "compression access method routine is NULL");
+
+ return result;
+}
+
+/* ----------
+ * attr_compression_options_are_equal
+ *
+ * Compare two attribute compression records
+ */
+static bool
+attr_compression_options_are_equal(Oid acoid1, Oid acoid2)
+{
+ CompressionAmOptions *a;
+ CompressionAmOptions *b;
+
+ a = lookup_compression_am_options(acoid1);
+ b = lookup_compression_am_options(acoid2);
+
+ if (a->amoid != b->amoid)
+ return false;
+
+ if (list_length(a->acoptions) != list_length(b->acoptions))
+ return false;
+
+ return equal(a->acoptions, b->acoptions);
+}
diff --git a/src/backend/access/index/amapi.c b/src/backend/access/index/amapi.c
index f395cb1ab4..353cd4bc39 100644
--- a/src/backend/access/index/amapi.c
+++ b/src/backend/access/index/amapi.c
@@ -17,6 +17,7 @@
#include "access/htup_details.h"
#include "catalog/pg_am.h"
#include "catalog/pg_opclass.h"
+#include "commands/defrem.h"
#include "utils/builtins.h"
#include "utils/syscache.h"
@@ -55,52 +56,12 @@ GetIndexAmRoutine(Oid amhandler)
IndexAmRoutine *
GetIndexAmRoutineByAmId(Oid amoid, bool noerror)
{
- HeapTuple tuple;
- Form_pg_am amform;
regproc amhandler;
/* Get handler function OID for the access method */
- tuple = SearchSysCache1(AMOID, ObjectIdGetDatum(amoid));
- if (!HeapTupleIsValid(tuple))
- {
- if (noerror)
- return NULL;
- elog(ERROR, "cache lookup failed for access method %u",
- amoid);
- }
- amform = (Form_pg_am) GETSTRUCT(tuple);
-
- /* Check if it's an index access method as opposed to some other AM */
- if (amform->amtype != AMTYPE_INDEX)
- {
- if (noerror)
- {
- ReleaseSysCache(tuple);
- return NULL;
- }
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("access method \"%s\" is not of type %s",
- NameStr(amform->amname), "INDEX")));
- }
-
- amhandler = amform->amhandler;
-
- /* Complain if handler OID is invalid */
- if (!RegProcedureIsValid(amhandler))
- {
- if (noerror)
- {
- ReleaseSysCache(tuple);
- return NULL;
- }
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("index access method \"%s\" does not have a handler",
- NameStr(amform->amname))));
- }
-
- ReleaseSysCache(tuple);
+ amhandler = get_am_handler_oid(amoid, AMTYPE_INDEX, noerror);
+ if (noerror && !RegProcedureIsValid(amhandler))
+ return NULL;
/* And finally, call the handler function to get the API struct. */
return GetIndexAmRoutine(amhandler);
diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c
index 7e34bee63e..d78ef2600c 100644
--- a/src/backend/bootstrap/bootstrap.c
+++ b/src/backend/bootstrap/bootstrap.c
@@ -733,6 +733,7 @@ DefineAttr(char *name, char *type, int attnum, int nullness)
attrtypes[attnum]->attcacheoff = -1;
attrtypes[attnum]->atttypmod = -1;
attrtypes[attnum]->attislocal = true;
+ attrtypes[attnum]->attcompression = InvalidOid;
if (nullness == BOOTCOL_NULL_FORCE_NOT_NULL)
{
diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index 0865240f11..5560fc05b8 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -34,7 +34,7 @@ CATALOG_HEADERS := \
pg_attrdef.h pg_constraint.h pg_inherits.h pg_index.h pg_operator.h \
pg_opfamily.h pg_opclass.h pg_am.h pg_amop.h pg_amproc.h \
pg_language.h pg_largeobject_metadata.h pg_largeobject.h pg_aggregate.h \
- pg_statistic_ext.h \
+ pg_statistic_ext.h pg_attr_compression.h \
pg_statistic.h pg_rewrite.h pg_trigger.h pg_event_trigger.h pg_description.h \
pg_cast.h pg_enum.h pg_namespace.h pg_conversion.h pg_depend.h \
pg_database.h pg_db_role_setting.h pg_tablespace.h pg_pltemplate.h \
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index 4f1d365357..61ebd316ba 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -23,6 +23,7 @@
#include "catalog/pg_am.h"
#include "catalog/pg_amop.h"
#include "catalog/pg_amproc.h"
+#include "catalog/pg_attr_compression.h"
#include "catalog/pg_attrdef.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_cast.h"
@@ -170,7 +171,8 @@ static const Oid object_classes[] = {
PublicationRelationId, /* OCLASS_PUBLICATION */
PublicationRelRelationId, /* OCLASS_PUBLICATION_REL */
SubscriptionRelationId, /* OCLASS_SUBSCRIPTION */
- TransformRelationId /* OCLASS_TRANSFORM */
+ TransformRelationId, /* OCLASS_TRANSFORM */
+ AttrCompressionRelationId /* OCLASS_ATTR_COMPRESSION */
};
@@ -1280,6 +1282,10 @@ doDeletion(const ObjectAddress *object, int flags)
DropTransformById(object->objectId);
break;
+ case OCLASS_ATTR_COMPRESSION:
+ RemoveAttributeCompression(object->objectId);
+ break;
+
/*
* These global object types are not supported here.
*/
@@ -2537,6 +2543,9 @@ getObjectClass(const ObjectAddress *object)
case TransformRelationId:
return OCLASS_TRANSFORM;
+
+ case AttrCompressionRelationId:
+ return OCLASS_ATTR_COMPRESSION;
}
/* shouldn't get here */
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index d223ba8537..a67755e40a 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -633,6 +633,7 @@ InsertPgAttributeTuple(Relation pg_attribute_rel,
values[Anum_pg_attribute_attislocal - 1] = BoolGetDatum(new_attribute->attislocal);
values[Anum_pg_attribute_attinhcount - 1] = Int32GetDatum(new_attribute->attinhcount);
values[Anum_pg_attribute_attcollation - 1] = ObjectIdGetDatum(new_attribute->attcollation);
+ values[Anum_pg_attribute_attcompression - 1] = ObjectIdGetDatum(new_attribute->attcompression);
/* start out with empty permissions and empty options */
nulls[Anum_pg_attribute_attacl - 1] = true;
@@ -1606,6 +1607,8 @@ RemoveAttributeById(Oid relid, AttrNumber attnum)
/* We don't want to keep stats for it anymore */
attStruct->attstattarget = 0;
+ attStruct->attcompression = InvalidOid;
+
/*
* Change the column name to something that isn't likely to conflict
*/
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 8b276bc430..e55a951f78 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -413,8 +413,9 @@ ConstructTupleDescriptor(Relation heapRelation,
to->attcacheoff = -1;
to->atttypmod = exprTypmod(indexkey);
to->attislocal = true;
+ to->attcompression = InvalidOid;
to->attcollation = (i < numkeyatts) ?
- collationObjectId[i] : InvalidOid;
+ collationObjectId[i] : InvalidOid;
ReleaseSysCache(tuple);
@@ -501,6 +502,7 @@ ConstructTupleDescriptor(Relation heapRelation,
to->attbyval = typeTup->typbyval;
to->attalign = typeTup->typalign;
to->attstorage = typeTup->typstorage;
+ to->attcompression = InvalidOid;
ReleaseSysCache(tuple);
}
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
index 7db942dcba..828240040b 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -23,6 +23,7 @@
#include "catalog/pg_am.h"
#include "catalog/pg_amop.h"
#include "catalog/pg_amproc.h"
+#include "catalog/pg_attr_compression.h"
#include "catalog/pg_attrdef.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_cast.h"
@@ -489,6 +490,18 @@ static const ObjectPropertyType ObjectProperty[] =
InvalidAttrNumber, /* no ACL (same as relation) */
OBJECT_STATISTIC_EXT,
true
+ },
+ {
+ AttrCompressionRelationId,
+ AttrCompressionIndexId,
+ ATTCOMPRESSIONOID,
+ -1,
+ InvalidAttrNumber,
+ InvalidAttrNumber,
+ InvalidAttrNumber,
+ InvalidAttrNumber,
+ -1,
+ false
}
};
@@ -3546,6 +3559,37 @@ getObjectDescription(const ObjectAddress *object)
break;
}
+ case OCLASS_ATTR_COMPRESSION:
+ {
+ char *attname;
+ HeapTuple tup;
+ Form_pg_attr_compression acform;
+
+ tup = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(object->objectId));
+ if (!HeapTupleIsValid(tup))
+ elog(ERROR, "cache lookup failed for attribute compression %u", object->objectId);
+
+ acform = (Form_pg_attr_compression) GETSTRUCT(tup);
+ if (OidIsValid(acform->acrelid))
+ {
+ appendStringInfo(&buffer, _("compression on "));
+ getRelationDescription(&buffer, acform->acrelid);
+
+ attname = get_attname(acform->acrelid, acform->acattnum, true);
+ if (attname)
+ {
+ appendStringInfo(&buffer, _(" column %s"), attname);
+ pfree(attname);
+ }
+ }
+ else
+ appendStringInfo(&buffer, _("attribute compression %u"), object->objectId);
+
+ ReleaseSysCache(tup);
+
+ break;
+ }
+
case OCLASS_TRANSFORM:
{
HeapTuple trfTup;
@@ -4070,6 +4114,10 @@ getObjectTypeDescription(const ObjectAddress *object)
appendStringInfoString(&buffer, "transform");
break;
+ case OCLASS_ATTR_COMPRESSION:
+ appendStringInfoString(&buffer, "attribute compression");
+ break;
+
/*
* There's intentionally no default: case here; we want the
* compiler to warn if a new OCLASS hasn't been handled above.
@@ -4614,6 +4662,15 @@ getObjectIdentityParts(const ObjectAddress *object,
break;
}
+ case OCLASS_ATTR_COMPRESSION:
+ {
+ appendStringInfo(&buffer, "%u",
+ object->objectId);
+ if (objname)
+ *objname = list_make1(psprintf("%u", object->objectId));
+ break;
+ }
+
case OCLASS_REWRITE:
{
Relation ruleDesc;
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index 4a6c99e090..e23abf64f1 100644
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -13,8 +13,8 @@ top_builddir = ../../..
include $(top_builddir)/src/Makefile.global
OBJS = amcmds.o aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o \
- collationcmds.o constraint.o conversioncmds.o copy.o createas.o \
- dbcommands.o define.o discard.o dropcmds.o \
+ collationcmds.o compressioncmds.o constraint.o conversioncmds.o copy.o \
+ createas.o dbcommands.o define.o discard.o dropcmds.o \
event_trigger.o explain.o extension.o foreigncmds.o functioncmds.o \
indexcmds.o lockcmds.o matview.o operatorcmds.o opclasscmds.o \
policy.o portalcmds.o prepare.o proclang.o publicationcmds.o \
diff --git a/src/backend/commands/alter.c b/src/backend/commands/alter.c
index eff325cc7d..eb49cfc54a 100644
--- a/src/backend/commands/alter.c
+++ b/src/backend/commands/alter.c
@@ -631,6 +631,7 @@ AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid,
case OCLASS_PUBLICATION_REL:
case OCLASS_SUBSCRIPTION:
case OCLASS_TRANSFORM:
+ case OCLASS_ATTR_COMPRESSION:
/* ignore object types that don't have schema-qualified names */
break;
diff --git a/src/backend/commands/amcmds.c b/src/backend/commands/amcmds.c
index f2173450ad..8f69f9e33b 100644
--- a/src/backend/commands/amcmds.c
+++ b/src/backend/commands/amcmds.c
@@ -175,6 +175,70 @@ get_am_type_oid(const char *amname, char amtype, bool missing_ok)
return oid;
}
+/*
+ * get_am_handler_oid - return access method handler Oid with additional
+ * type checking.
+ *
+ * If noerror is false, throw an error if access method not found or its
+ * type not equal to specified type, or its handler is not valid.
+ *
+ * If amtype is not '\0', an error is raised if the AM found is not of the
+ * given type.
+ */
+regproc
+get_am_handler_oid(Oid amOid, char amtype, bool noerror)
+{
+ HeapTuple tup;
+ Form_pg_am amform;
+ regproc amhandler;
+
+ /* Get handler function OID for the access method */
+ tup = SearchSysCache1(AMOID, ObjectIdGetDatum(amOid));
+ if (!HeapTupleIsValid(tup))
+ {
+ if (noerror)
+ return InvalidOid;
+
+ elog(ERROR, "cache lookup failed for access method %u", amOid);
+ }
+
+ amform = (Form_pg_am) GETSTRUCT(tup);
+ amhandler = amform->amhandler;
+
+ if (amtype != '\0' && amform->amtype != amtype)
+ {
+ if (noerror)
+ {
+ ReleaseSysCache(tup);
+ return InvalidOid;
+ }
+
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("access method \"%s\" is not of type %s",
+ NameStr(amform->amname),
+ get_am_type_string(amtype))));
+ }
+
+ /* Complain if handler OID is invalid */
+ if (!RegProcedureIsValid(amhandler))
+ {
+ if (noerror)
+ {
+ ReleaseSysCache(tup);
+ return InvalidOid;
+ }
+
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("access method \"%s\" does not have a handler",
+ NameStr(amform->amname))));
+ }
+
+ ReleaseSysCache(tup);
+ return amhandler;
+}
+
/*
* get_index_am_oid - given an access method name, look up its OID
* and verify it corresponds to an index AM.
@@ -185,6 +249,16 @@ get_index_am_oid(const char *amname, bool missing_ok)
return get_am_type_oid(amname, AMTYPE_INDEX, missing_ok);
}
+/*
+ * get_index_am_oid - given an access method name, look up its OID
+ * and verify it corresponds to an index AM.
+ */
+Oid
+get_compression_am_oid(const char *amname, bool missing_ok)
+{
+ return get_am_type_oid(amname, AMTYPE_COMPRESSION, missing_ok);
+}
+
/*
* get_am_oid - given an access method name, look up its OID.
* The type is not checked.
@@ -225,6 +299,8 @@ get_am_type_string(char amtype)
{
case AMTYPE_INDEX:
return "INDEX";
+ case AMTYPE_COMPRESSION:
+ return "COMPRESSION";
default:
/* shouldn't happen */
elog(ERROR, "invalid access method type '%c'", amtype);
@@ -263,6 +339,14 @@ lookup_index_am_handler_func(List *handler_name, char amtype)
NameListToString(handler_name),
"index_am_handler")));
break;
+ case AMTYPE_COMPRESSION:
+ if (get_func_rettype(handlerOid) != COMPRESSION_AM_HANDLEROID)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("function %s must return type %s",
+ NameListToString(handler_name),
+ "compression_am_handler")));
+ break;
default:
elog(ERROR, "unrecognized access method type \"%c\"", amtype);
}
diff --git a/src/backend/commands/compressioncmds.c b/src/backend/commands/compressioncmds.c
new file mode 100644
index 0000000000..3768a38459
--- /dev/null
+++ b/src/backend/commands/compressioncmds.c
@@ -0,0 +1,653 @@
+/*-------------------------------------------------------------------------
+ *
+ * compressioncmds.c
+ * Routines for SQL commands for compression access methods
+ *
+ * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ * src/backend/commands/compressioncmds.c
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+#include "miscadmin.h"
+
+#include "access/cmapi.h"
+#include "access/heapam.h"
+#include "access/htup_details.h"
+#include "access/reloptions.h"
+#include "access/xact.h"
+#include "catalog/catalog.h"
+#include "catalog/dependency.h"
+#include "catalog/indexing.h"
+#include "catalog/pg_attr_compression.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_depend.h"
+#include "catalog/pg_proc.h"
+#include "catalog/pg_type.h"
+#include "commands/defrem.h"
+#include "parser/parse_func.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/syscache.h"
+#include "utils/snapmgr.h"
+
+/*
+ * When conditions of compression satisfies one if builtin attribute
+ * compresssion tuples the compressed attribute will be linked to
+ * builtin compression without new record in pg_attr_compression.
+ * So the fact that the column has a builtin compression we only can find out
+ * by its dependency.
+ */
+static void
+lookup_builtin_dependencies(Oid attrelid, AttrNumber attnum,
+ List **amoids)
+{
+ LOCKMODE lock = AccessShareLock;
+ Oid amoid = InvalidOid;
+ HeapTuple tup;
+ Relation rel;
+ SysScanDesc scan;
+ ScanKeyData key[3];
+
+ rel = heap_open(DependRelationId, lock);
+
+ ScanKeyInit(&key[0],
+ Anum_pg_depend_classid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationRelationId));
+ ScanKeyInit(&key[1],
+ Anum_pg_depend_objid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(attrelid));
+ ScanKeyInit(&key[2],
+ Anum_pg_depend_objsubid,
+ BTEqualStrategyNumber, F_INT4EQ,
+ Int32GetDatum((int32) attnum));
+
+ scan = systable_beginscan(rel, DependDependerIndexId, true,
+ NULL, 3, key);
+
+ while (HeapTupleIsValid(tup = systable_getnext(scan)))
+ {
+ Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tup);
+
+ if (depform->refclassid == AttrCompressionRelationId)
+ {
+ Assert(IsBuiltinCompression(depform->refobjid));
+ amoid = GetAttrCompressionAmOid(depform->refobjid);
+ *amoids = list_append_unique_oid(*amoids, amoid);
+ }
+ }
+
+ systable_endscan(scan);
+ heap_close(rel, lock);
+}
+
+/*
+ * Find identical attribute compression for reuse and fill the list with
+ * used compression access methods.
+ */
+static Oid
+lookup_attribute_compression(Oid attrelid, AttrNumber attnum,
+ Oid amoid, Datum acoptions, List **previous_amoids)
+{
+ Relation rel;
+ HeapTuple tuple;
+ SysScanDesc scan;
+ FmgrInfo arrayeq_info;
+ Oid result = InvalidOid;
+ ScanKeyData key[2];
+
+ /* fill FmgrInfo for array_eq function */
+ fmgr_info(F_ARRAY_EQ, &arrayeq_info);
+
+ Assert((attrelid > 0 && attnum > 0) || (attrelid == 0 && attnum == 0));
+ if (previous_amoids)
+ lookup_builtin_dependencies(attrelid, attnum, previous_amoids);
+
+ rel = heap_open(AttrCompressionRelationId, AccessShareLock);
+ ScanKeyInit(&key[0],
+ Anum_pg_attr_compression_acrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(attrelid));
+
+ ScanKeyInit(&key[1],
+ Anum_pg_attr_compression_acattnum,
+ BTEqualStrategyNumber, F_INT2EQ,
+ Int16GetDatum(attnum));
+
+ scan = systable_beginscan(rel, AttrCompressionRelidAttnumIndexId,
+ true, NULL, 2, key);
+ while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+ {
+ Oid acoid,
+ tup_amoid;
+ Datum values[Natts_pg_attr_compression];
+ bool nulls[Natts_pg_attr_compression];
+
+ heap_deform_tuple(tuple, RelationGetDescr(rel), values, nulls);
+ acoid = DatumGetObjectId(values[Anum_pg_attr_compression_acoid - 1]);
+ tup_amoid = get_am_oid(
+ NameStr(*DatumGetName(values[Anum_pg_attr_compression_acname - 1])), false);
+
+ if (previous_amoids)
+ *previous_amoids = list_append_unique_oid(*previous_amoids, tup_amoid);
+
+ if (tup_amoid != amoid)
+ continue;
+
+ /*
+ * even if we found the match, we still need to acquire all previous
+ * access methods so don't break cycle if previous_amoids is not NULL
+ */
+ if (nulls[Anum_pg_attr_compression_acoptions - 1])
+ {
+ if (DatumGetPointer(acoptions) == NULL)
+ result = acoid;
+ }
+ else
+ {
+ bool equal;
+
+ /* check if arrays for WITH options are equal */
+ equal = DatumGetBool(CallerFInfoFunctionCall2(
+ array_eq,
+ &arrayeq_info,
+ InvalidOid,
+ acoptions,
+ values[Anum_pg_attr_compression_acoptions - 1]));
+ if (equal)
+ result = acoid;
+ }
+
+ if (previous_amoids == NULL && OidIsValid(result))
+ break;
+ }
+
+ systable_endscan(scan);
+ heap_close(rel, AccessShareLock);
+ return result;
+}
+
+/*
+ * Link compression with an attribute. Creates a row in pg_attr_compression
+ * if needed.
+ *
+ * When compression is not specified returns default attribute compression.
+ * It is possible case for CREATE TABLE and ADD COLUMN commands
+ * where COMPRESSION syntax is optional.
+ *
+ * If any of builtin attribute compression tuples satisfies conditions
+ * returns it.
+ *
+ * For ALTER command check for previous attribute compression record with
+ * identical compression options and reuse it if found any.
+ *
+ * Note we create attribute compression for EXTERNAL storage too, so when
+ * storage is changed we can start compression on future tuples right away.
+ */
+Oid
+CreateAttributeCompression(Form_pg_attribute att,
+ ColumnCompression *compression,
+ bool *need_rewrite, List **preserved_amoids)
+{
+ Relation rel;
+ HeapTuple newtup;
+ Oid acoid = InvalidOid,
+ amoid;
+ regproc amhandler;
+ Datum arropt;
+ Datum values[Natts_pg_attr_compression];
+ bool nulls[Natts_pg_attr_compression];
+
+ ObjectAddress myself,
+ amref;
+
+ CompressionAmRoutine *routine;
+
+ /* No compression for PLAIN storage. */
+ if (att->attstorage == 'p')
+ return InvalidOid;
+
+ /* Fallback to default compression if it's not specified */
+ if (compression == NULL)
+ return DefaultCompressionOid;
+
+ amoid = get_compression_am_oid(compression->amname, false);
+ if (compression->options)
+ arropt = optionListToArray(compression->options, true);
+ else
+ arropt = PointerGetDatum(NULL);
+
+ /* Try to find builtin compression first */
+ acoid = lookup_attribute_compression(0, 0, amoid, arropt, NULL);
+
+ /*
+ * attrelid will be invalid on CREATE TABLE, no need for table rewrite
+ * check.
+ */
+ if (OidIsValid(att->attrelid))
+ {
+ Oid attacoid;
+ List *previous_amoids = NIL;
+
+ /*
+ * Try to find identical compression from previous tuples, and fill
+ * the list of previous compresssion methods
+ */
+ attacoid = lookup_attribute_compression(att->attrelid, att->attnum,
+ amoid, arropt, &previous_amoids);
+ if (!OidIsValid(acoid))
+ acoid = attacoid;
+
+ /*
+ * Determine if the column needs rewrite or not. Rewrite conditions: -
+ * SET COMPRESSION without PRESERVE - SET COMPRESSION with PRESERVE
+ * but not with full list of previous access methods.
+ */
+ if (need_rewrite != NULL)
+ {
+ /* no rewrite by default */
+ *need_rewrite = false;
+
+ Assert(preserved_amoids != NULL);
+
+ if (compression->preserve == NIL)
+ {
+ Assert(!IsBinaryUpgrade);
+ *need_rewrite = true;
+ }
+ else
+ {
+ ListCell *cell;
+
+ foreach(cell, compression->preserve)
+ {
+ char *amname_p = strVal(lfirst(cell));
+ Oid amoid_p = get_am_oid(amname_p, false);
+
+ if (!list_member_oid(previous_amoids, amoid_p))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("\"%s\" compression access method cannot be preserved", amname_p),
+ errhint("use \"pg_column_compression\" function for list of compression methods")
+ ));
+
+ *preserved_amoids = list_append_unique_oid(*preserved_amoids, amoid_p);
+
+ /*
+ * Remove from previous list, also protect from multiple
+ * mentions of one access method in PRESERVE list
+ */
+ previous_amoids = list_delete_oid(previous_amoids, amoid_p);
+ }
+
+ /*
+ * If the list of previous Oids is not empty after deletions
+ * then we need to rewrite tuples in the table.
+ *
+ * In binary upgrade list will not be free since it contains
+ * Oid of builtin compression access method.
+ */
+ if (!IsBinaryUpgrade && list_length(previous_amoids) != 0)
+ *need_rewrite = true;
+ }
+ }
+
+ /* Cleanup */
+ list_free(previous_amoids);
+ }
+
+ if (IsBinaryUpgrade && !OidIsValid(acoid))
+ elog(ERROR, "could not restore attribute compression data");
+
+ /* Return Oid if we already found identical compression on this column */
+ if (OidIsValid(acoid))
+ {
+ if (DatumGetPointer(arropt) != NULL)
+ pfree(DatumGetPointer(arropt));
+
+ return acoid;
+ }
+
+ /* Initialize buffers for new tuple values */
+ memset(values, 0, sizeof(values));
+ memset(nulls, false, sizeof(nulls));
+
+ amhandler = get_am_handler_oid(amoid, AMTYPE_COMPRESSION, false);
+
+ rel = heap_open(AttrCompressionRelationId, RowExclusiveLock);
+
+ acoid = GetNewOidWithIndex(rel, AttrCompressionIndexId,
+ Anum_pg_attr_compression_acoid);
+ if (acoid < FirstNormalObjectId)
+ {
+ /* this is database initialization */
+ heap_close(rel, RowExclusiveLock);
+ return DefaultCompressionOid;
+ }
+
+ /* we need routine only to call cmcheck function */
+ routine = InvokeCompressionAmHandler(amhandler);
+ if (routine->cmcheck != NULL)
+ routine->cmcheck(att, compression->options);
+ pfree(routine);
+
+ values[Anum_pg_attr_compression_acoid - 1] = ObjectIdGetDatum(acoid);
+ values[Anum_pg_attr_compression_acname - 1] = CStringGetDatum(compression->amname);
+ values[Anum_pg_attr_compression_acrelid - 1] = ObjectIdGetDatum(att->attrelid);
+ values[Anum_pg_attr_compression_acattnum - 1] = Int32GetDatum(att->attnum);
+
+ if (compression->options)
+ values[Anum_pg_attr_compression_acoptions - 1] = arropt;
+ else
+ nulls[Anum_pg_attr_compression_acoptions - 1] = true;
+
+ newtup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
+ CatalogTupleInsert(rel, newtup);
+ heap_freetuple(newtup);
+ heap_close(rel, RowExclusiveLock);
+
+ ObjectAddressSet(myself, AttrCompressionRelationId, acoid);
+ ObjectAddressSet(amref, AccessMethodRelationId, amoid);
+
+ recordDependencyOn(&myself, &amref, DEPENDENCY_NORMAL);
+ recordDependencyOnCurrentExtension(&myself, false);
+
+ /* Make the changes visible */
+ CommandCounterIncrement();
+ return acoid;
+}
+
+/*
+ * Remove the attribute compression record from pg_attr_compression.
+ */
+void
+RemoveAttributeCompression(Oid acoid)
+{
+ Relation relation;
+ HeapTuple tup;
+ Form_pg_attr_compression acform;
+
+ tup = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+ if (!HeapTupleIsValid(tup))
+ elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+ acform = (Form_pg_attr_compression) GETSTRUCT(tup);
+
+ /* check we're not trying to remove builtin attribute compression */
+ Assert(acform->acrelid != 0);
+
+ /* delete the record from catalogs */
+ relation = heap_open(AttrCompressionRelationId, RowExclusiveLock);
+ CatalogTupleDelete(relation, &tup->t_self);
+ heap_close(relation, RowExclusiveLock);
+ ReleaseSysCache(tup);
+}
+
+/*
+ * CleanupAttributeCompression
+ *
+ * Remove entries in pg_attr_compression except current attribute compression
+ * and related with specified list of access methods.
+ */
+void
+CleanupAttributeCompression(Oid relid, AttrNumber attnum, List *keepAmOids)
+{
+ Oid acoid,
+ amoid;
+ Relation rel;
+ SysScanDesc scan;
+ ScanKeyData key[3];
+ HeapTuple tuple,
+ attrtuple;
+ Form_pg_attribute attform;
+ List *removed = NIL;
+ ListCell *lc;
+
+ attrtuple = SearchSysCache2(ATTNUM,
+ ObjectIdGetDatum(relid),
+ Int16GetDatum(attnum));
+
+ if (!HeapTupleIsValid(attrtuple))
+ elog(ERROR, "cache lookup failed for attribute %d of relation %u",
+ attnum, relid);
+ attform = (Form_pg_attribute) GETSTRUCT(attrtuple);
+ acoid = attform->attcompression;
+ ReleaseSysCache(attrtuple);
+
+ Assert(relid > 0 && attnum > 0);
+
+ if (IsBinaryUpgrade)
+ goto builtin_removal;
+
+ rel = heap_open(AttrCompressionRelationId, RowExclusiveLock);
+
+ ScanKeyInit(&key[0],
+ Anum_pg_attr_compression_acrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(relid));
+
+ ScanKeyInit(&key[1],
+ Anum_pg_attr_compression_acattnum,
+ BTEqualStrategyNumber, F_INT2EQ,
+ Int16GetDatum(attnum));
+
+ scan = systable_beginscan(rel, AttrCompressionRelidAttnumIndexId,
+ true, NULL, 2, key);
+
+ /* Remove attribute compression tuples and collect removed Oids to list */
+ while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+ {
+ Form_pg_attr_compression acform;
+
+ acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+ amoid = get_am_oid(NameStr(acform->acname), false);
+
+ /* skip current compression */
+ if (acform->acoid == acoid)
+ continue;
+
+ if (!list_member_oid(keepAmOids, amoid))
+ {
+ removed = lappend_oid(removed, acform->acoid);
+ CatalogTupleDelete(rel, &tuple->t_self);
+ }
+ }
+
+ systable_endscan(scan);
+ heap_close(rel, RowExclusiveLock);
+
+ /* Now remove dependencies */
+ rel = heap_open(DependRelationId, RowExclusiveLock);
+ foreach(lc, removed)
+ {
+ Oid tup_acoid = lfirst_oid(lc);
+
+ ScanKeyInit(&key[0],
+ Anum_pg_depend_classid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(AttrCompressionRelationId));
+ ScanKeyInit(&key[1],
+ Anum_pg_depend_objid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(tup_acoid));
+
+ scan = systable_beginscan(rel, DependDependerIndexId, true,
+ NULL, 2, key);
+
+ while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+ CatalogTupleDelete(rel, &tuple->t_self);
+
+ systable_endscan(scan);
+ }
+ heap_close(rel, RowExclusiveLock);
+
+builtin_removal:
+ /* Now remove dependencies with builtin compressions */
+ rel = heap_open(DependRelationId, RowExclusiveLock);
+ ScanKeyInit(&key[0],
+ Anum_pg_depend_classid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationRelationId));
+ ScanKeyInit(&key[1],
+ Anum_pg_depend_objid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(relid));
+ ScanKeyInit(&key[2],
+ Anum_pg_depend_objsubid,
+ BTEqualStrategyNumber, F_INT4EQ,
+ Int32GetDatum((int32) attnum));
+
+ scan = systable_beginscan(rel, DependDependerIndexId, true,
+ NULL, 3, key);
+
+ while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+ {
+ Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tuple);
+
+ if (depform->refclassid != AttrCompressionRelationId)
+ continue;
+
+ /* skip current compression */
+ if (depform->refobjid == acoid)
+ continue;
+
+ amoid = GetAttrCompressionAmOid(depform->refobjid);
+ if (!list_member_oid(keepAmOids, amoid))
+ CatalogTupleDelete(rel, &tuple->t_self);
+ }
+
+ systable_endscan(scan);
+ heap_close(rel, RowExclusiveLock);
+}
+
+/*
+ * Construct ColumnCompression node by attribute compression Oid.
+ */
+ColumnCompression *
+MakeColumnCompression(Oid acoid)
+{
+ HeapTuple tuple;
+ Form_pg_attr_compression acform;
+ ColumnCompression *node;
+
+ if (!OidIsValid(acoid))
+ return NULL;
+
+ tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+ acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+ node = makeNode(ColumnCompression);
+ node->amname = pstrdup(NameStr(acform->acname));
+ ReleaseSysCache(tuple);
+
+ /*
+ * fill attribute compression options too. We could've do it above but
+ * it's easier to call this helper.
+ */
+ node->options = GetAttrCompressionOptions(acoid);
+
+ return node;
+}
+
+/*
+ * Compare compression options for two columns.
+ */
+void
+CheckCompressionMismatch(ColumnCompression *c1, ColumnCompression *c2,
+ const char *attributeName)
+{
+ if (strcmp(c1->amname, c2->amname))
+ ereport(ERROR,
+ (errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("column \"%s\" has a compression method conflict",
+ attributeName),
+ errdetail("%s versus %s", c1->amname, c2->amname)));
+
+ if (!equal(c1->options, c2->options))
+ ereport(ERROR,
+ (errcode(ERRCODE_DATATYPE_MISMATCH),
+ errmsg("column \"%s\" has a compression options conflict",
+ attributeName),
+ errdetail("(%s) versus (%s)",
+ formatRelOptions(c1->options),
+ formatRelOptions(c2->options))));
+}
+
+/*
+ * Return list of compression methods used in specified column.
+ */
+Datum
+pg_column_compression(PG_FUNCTION_ARGS)
+{
+ Oid relOid = PG_GETARG_OID(0);
+ char *attname = TextDatumGetCString(PG_GETARG_TEXT_P(1));
+ Relation rel;
+ HeapTuple tuple;
+ AttrNumber attnum;
+ List *amoids = NIL;
+ Oid amoid;
+ ListCell *lc;
+
+ ScanKeyData key[2];
+ SysScanDesc scan;
+ StringInfoData result;
+
+ attnum = get_attnum(relOid, attname);
+ if (attnum == InvalidAttrNumber)
+ PG_RETURN_NULL();
+
+ /* Collect related builtin compression access methods */
+ lookup_builtin_dependencies(relOid, attnum, &amoids);
+
+ /* Collect other related access methods */
+ rel = heap_open(AttrCompressionRelationId, AccessShareLock);
+
+ ScanKeyInit(&key[0],
+ Anum_pg_attr_compression_acrelid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(relOid));
+ ScanKeyInit(&key[1],
+ Anum_pg_attr_compression_acattnum,
+ BTEqualStrategyNumber, F_INT2EQ,
+ Int16GetDatum(attnum));
+
+ scan = systable_beginscan(rel, AttrCompressionRelidAttnumIndexId,
+ true, NULL, 2, key);
+ while (HeapTupleIsValid(tuple = systable_getnext(scan)))
+ {
+ Form_pg_attr_compression acform;
+
+ acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+ amoid = get_am_oid(NameStr(acform->acname), false);
+ amoids = list_append_unique_oid(amoids, amoid);
+ }
+
+ systable_endscan(scan);
+ heap_close(rel, AccessShareLock);
+
+ if (!list_length(amoids))
+ PG_RETURN_NULL();
+
+ /* Construct the list separated by comma */
+ amoid = InvalidOid;
+ initStringInfo(&result);
+ foreach(lc, amoids)
+ {
+ if (OidIsValid(amoid))
+ appendStringInfoString(&result, ", ");
+
+ amoid = lfirst_oid(lc);
+ appendStringInfoString(&result, get_am_name(amoid));
+ }
+
+ PG_RETURN_TEXT_P(CStringGetTextDatum(result.data));
+}
diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c
index eecc85d14e..ee67d07b2f 100644
--- a/src/backend/commands/event_trigger.c
+++ b/src/backend/commands/event_trigger.c
@@ -1209,6 +1209,7 @@ EventTriggerSupportsObjectClass(ObjectClass objclass)
case OCLASS_PUBLICATION_REL:
case OCLASS_SUBSCRIPTION:
case OCLASS_TRANSFORM:
+ case OCLASS_ATTR_COMPRESSION:
return true;
/*
diff --git a/src/backend/commands/foreigncmds.c b/src/backend/commands/foreigncmds.c
index 5c53aeeaeb..6db6e38a07 100644
--- a/src/backend/commands/foreigncmds.c
+++ b/src/backend/commands/foreigncmds.c
@@ -48,50 +48,6 @@ typedef struct
/* Internal functions */
static void import_error_callback(void *arg);
-
-/*
- * Convert a DefElem list to the text array format that is used in
- * pg_foreign_data_wrapper, pg_foreign_server, pg_user_mapping, and
- * pg_foreign_table.
- *
- * Returns the array in the form of a Datum, or PointerGetDatum(NULL)
- * if the list is empty.
- *
- * Note: The array is usually stored to database without further
- * processing, hence any validation should be done before this
- * conversion.
- */
-static Datum
-optionListToArray(List *options)
-{
- ArrayBuildState *astate = NULL;
- ListCell *cell;
-
- foreach(cell, options)
- {
- DefElem *def = lfirst(cell);
- const char *value;
- Size len;
- text *t;
-
- value = defGetString(def);
- len = VARHDRSZ + strlen(def->defname) + 1 + strlen(value);
- t = palloc(len + 1);
- SET_VARSIZE(t, len);
- sprintf(VARDATA(t), "%s=%s", def->defname, value);
-
- astate = accumArrayResult(astate, PointerGetDatum(t),
- false, TEXTOID,
- CurrentMemoryContext);
- }
-
- if (astate)
- return makeArrayResult(astate, CurrentMemoryContext);
-
- return PointerGetDatum(NULL);
-}
-
-
/*
* Transform a list of DefElem into text array format. This is substantially
* the same thing as optionListToArray(), except we recognize SET/ADD/DROP
@@ -178,7 +134,7 @@ transformGenericOptions(Oid catalogId,
}
}
- result = optionListToArray(resultOptions);
+ result = optionListToArray(resultOptions, false);
if (OidIsValid(fdwvalidator))
{
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 7c0cf0d7ee..0c7c201952 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -22,6 +22,7 @@
#include "access/relscan.h"
#include "access/sysattr.h"
#include "access/tupconvert.h"
+#include "access/tuptoaster.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "catalog/catalog.h"
@@ -33,6 +34,7 @@
#include "catalog/objectaccess.h"
#include "catalog/partition.h"
#include "catalog/pg_am.h"
+#include "catalog/pg_attr_compression.h"
#include "catalog/pg_collation.h"
#include "catalog/pg_constraint.h"
#include "catalog/pg_depend.h"
@@ -40,6 +42,7 @@
#include "catalog/pg_inherits.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_opclass.h"
+#include "catalog/pg_proc.h"
#include "catalog/pg_tablespace.h"
#include "catalog/pg_trigger.h"
#include "catalog/pg_type.h"
@@ -89,6 +92,7 @@
#include "storage/smgr.h"
#include "utils/acl.h"
#include "utils/builtins.h"
+#include "utils/datum.h"
#include "utils/fmgroids.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
@@ -367,6 +371,8 @@ static bool check_for_column_name_collision(Relation rel, const char *colname,
bool if_not_exists);
static void add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid);
static void add_column_collation_dependency(Oid relid, int32 attnum, Oid collid);
+static void add_column_compression_dependency(Oid relid, int32 attnum, Oid acoid);
+static void set_column_compression_relid(Oid acoid, Oid relid);
static void ATPrepAddOids(List **wqueue, Relation rel, bool recurse,
AlterTableCmd *cmd, LOCKMODE lockmode);
static void ATPrepDropNotNull(Relation rel, bool recurse, bool recursing);
@@ -465,6 +471,8 @@ static void ATExecGenericOptions(Relation rel, List *options);
static void ATExecEnableRowSecurity(Relation rel);
static void ATExecDisableRowSecurity(Relation rel);
static void ATExecForceNoForceRowSecurity(Relation rel, bool force_rls);
+static ObjectAddress ATExecSetCompression(AlteredTableInfo *tab, Relation rel,
+ const char *column, ColumnCompression *compression, LOCKMODE lockmode);
static void copy_relation_data(SMgrRelation rel, SMgrRelation dst,
ForkNumber forkNum, char relpersistence);
@@ -519,6 +527,7 @@ ObjectAddress
DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
ObjectAddress *typaddress, const char *queryString)
{
+ int i;
char relname[NAMEDATALEN];
Oid namespaceId;
Oid relationId;
@@ -693,6 +702,8 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
* We can set the atthasdef flags now in the tuple descriptor; this just
* saves StoreAttrDefault from having to do an immediate update of the
* pg_attribute rows.
+ *
+ * Also create attribute compression records if needed.
*/
rawDefaults = NIL;
cookedDefaults = NIL;
@@ -739,6 +750,13 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
if (colDef->identity)
attr->attidentity = colDef->identity;
+
+ if (relkind == RELKIND_RELATION || relkind == RELKIND_PARTITIONED_TABLE)
+ attr->attcompression = CreateAttributeCompression(attr,
+ colDef->compression,
+ NULL, NULL);
+ else
+ attr->attcompression = InvalidOid;
}
/*
@@ -787,6 +805,23 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId,
*/
rel = relation_open(relationId, AccessExclusiveLock);
+ /*
+ * Specify relation Oid in attribute compression tuples, and create
+ * dependencies.
+ */
+ for (i = 0; i < (RelationGetDescr(rel))->natts; i++)
+ {
+ Form_pg_attribute attr;
+
+ attr = TupleDescAttr(RelationGetDescr(rel), i);
+ if (OidIsValid(attr->attcompression))
+ {
+ set_column_compression_relid(attr->attcompression, relationId);
+ add_column_compression_dependency(attr->attrelid, attr->attnum,
+ attr->attcompression);
+ }
+ }
+
/* Process and store partition bound, if any. */
if (stmt->partbound)
{
@@ -2117,6 +2152,19 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
storage_name(def->storage),
storage_name(attribute->attstorage))));
+ if (OidIsValid(attribute->attcompression))
+ {
+ ColumnCompression *compression =
+ MakeColumnCompression(attribute->attcompression);
+
+ if (!def->compression)
+ def->compression = compression;
+ else
+ CheckCompressionMismatch(def->compression,
+ compression,
+ attributeName);
+ }
+
def->inhcount++;
/* Merge of NOT NULL constraints = OR 'em together */
def->is_not_null |= attribute->attnotnull;
@@ -2144,6 +2192,7 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
def->collOid = attribute->attcollation;
def->constraints = NIL;
def->location = -1;
+ def->compression = MakeColumnCompression(attribute->attcompression);
inhSchema = lappend(inhSchema, def);
newattno[parent_attno - 1] = ++child_attno;
}
@@ -2353,6 +2402,13 @@ MergeAttributes(List *schema, List *supers, char relpersistence,
storage_name(def->storage),
storage_name(newdef->storage))));
+ if (!def->compression)
+ def->compression = newdef->compression;
+ else if (newdef->compression)
+ CheckCompressionMismatch(def->compression,
+ newdef->compression,
+ attributeName);
+
/* Mark the column as locally defined */
def->is_local = true;
/* Merge of NOT NULL constraints = OR 'em together */
@@ -3440,6 +3496,7 @@ AlterTableGetLockLevel(List *cmds)
*/
case AT_GenericOptions:
case AT_AlterColumnGenericOptions:
+ case AT_SetCompression:
cmd_lockmode = AccessExclusiveLock;
break;
@@ -3915,6 +3972,7 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd,
case AT_DisableRowSecurity:
case AT_ForceRowSecurity:
case AT_NoForceRowSecurity:
+ case AT_SetCompression:
ATSimplePermissions(rel, ATT_TABLE);
/* These commands never recurse */
/* No command-specific prep needed */
@@ -4289,6 +4347,11 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel,
Assert(rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
ATExecDetachPartition(rel, ((PartitionCmd *) cmd->def)->name);
break;
+ case AT_SetCompression:
+ address = ATExecSetCompression(tab, rel, cmd->name,
+ (ColumnCompression *) cmd->def,
+ lockmode);
+ break;
default: /* oops */
elog(ERROR, "unrecognized alter table type: %d",
(int) cmd->subtype);
@@ -4354,8 +4417,9 @@ ATRewriteTables(AlterTableStmt *parsetree, List **wqueue, LOCKMODE lockmode)
/*
* We only need to rewrite the table if at least one column needs to
- * be recomputed, we are adding/removing the OID column, or we are
- * changing its persistence.
+ * be recomputed, we are adding/removing the OID column, we are
+ * changing its persistence, or we are changing compression type of a
+ * column without preserving old ones.
*
* There are two reasons for requiring a rewrite when changing
* persistence: on one hand, we need to ensure that the buffers
@@ -5517,6 +5581,16 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
attribute.attislocal = colDef->is_local;
attribute.attinhcount = colDef->inhcount;
attribute.attcollation = collOid;
+
+ /* create attribute compresssion record */
+ if (rel->rd_rel->relkind == RELKIND_RELATION ||
+ rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+ attribute.attcompression = CreateAttributeCompression(&attribute,
+ colDef->compression,
+ NULL, NULL);
+ else
+ attribute.attcompression = InvalidOid;
+
/* attribute.attacl is handled by InsertPgAttributeTuple */
ReleaseSysCache(typeTuple);
@@ -5691,6 +5765,7 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,
*/
add_column_datatype_dependency(myrelid, newattnum, attribute.atttypid);
add_column_collation_dependency(myrelid, newattnum, attribute.attcollation);
+ add_column_compression_dependency(myrelid, newattnum, attribute.attcompression);
/*
* Propagate to children as appropriate. Unlike most other ALTER
@@ -5813,6 +5888,30 @@ add_column_datatype_dependency(Oid relid, int32 attnum, Oid typid)
recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
}
+/* Update acrelid in pg_attr_compression */
+static void
+set_column_compression_relid(Oid acoid, Oid relid)
+{
+ Relation rel;
+ HeapTuple tuple;
+ Form_pg_attr_compression acform;
+
+ Assert(OidIsValid(relid));
+ if (IsBuiltinCompression(acoid))
+ return;
+
+ rel = heap_open(AttrCompressionRelationId, RowExclusiveLock);
+ tuple = SearchSysCache1(ATTCOMPRESSIONOID, ObjectIdGetDatum(acoid));
+ if (!HeapTupleIsValid(tuple))
+ elog(ERROR, "cache lookup failed for attribute compression %u", acoid);
+
+ acform = (Form_pg_attr_compression) GETSTRUCT(tuple);
+ acform->acrelid = relid;
+ CatalogTupleUpdate(rel, &tuple->t_self, tuple);
+ ReleaseSysCache(tuple);
+ heap_close(rel, RowExclusiveLock);
+}
+
/*
* Install a column's dependency on its collation.
*/
@@ -5835,6 +5934,106 @@ add_column_collation_dependency(Oid relid, int32 attnum, Oid collid)
}
}
+/*
+ * Install a dependency for compression options on its column.
+ *
+ * If it is builtin attribute compression then column depends on it. This
+ * is used to determine connection between column and builtin attribute
+ * compression in ALTER SET COMPRESSION command.
+ *
+ * If dependency is already there the whole thing is skipped.
+ */
+static void
+add_column_compression_dependency(Oid relid, int32 attnum, Oid acoid)
+{
+ bool found = false;
+ Relation depRel;
+ ObjectAddress acref,
+ attref;
+ HeapTuple depTup;
+ ScanKeyData key[3];
+ SysScanDesc scan;
+
+ if (!OidIsValid(acoid))
+ return;
+
+ Assert(relid > 0 && attnum > 0);
+ ObjectAddressSet(acref, AttrCompressionRelationId, acoid);
+ ObjectAddressSubSet(attref, RelationRelationId, relid, attnum);
+
+ depRel = heap_open(DependRelationId, AccessShareLock);
+
+ if (IsBuiltinCompression(acoid))
+ {
+ ScanKeyInit(&key[0],
+ Anum_pg_depend_classid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationRelationId));
+ ScanKeyInit(&key[1],
+ Anum_pg_depend_objid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(relid));
+ ScanKeyInit(&key[2],
+ Anum_pg_depend_objsubid,
+ BTEqualStrategyNumber, F_INT4EQ,
+ Int32GetDatum((int32) attnum));
+
+ scan = systable_beginscan(depRel, DependDependerIndexId, true,
+ NULL, 3, key);
+
+ while (HeapTupleIsValid(depTup = systable_getnext(scan)))
+ {
+ Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(depTup);
+
+ if (foundDep->refclassid == AttrCompressionRelationId &&
+ foundDep->refobjid == acoid)
+ {
+ found = true;
+ break;
+ }
+ }
+ systable_endscan(scan);
+
+ if (!found)
+ recordDependencyOn(&attref, &acref, DEPENDENCY_NORMAL);
+ }
+ else
+ {
+ ScanKeyInit(&key[0],
+ Anum_pg_depend_refclassid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(RelationRelationId));
+ ScanKeyInit(&key[1],
+ Anum_pg_depend_refobjid,
+ BTEqualStrategyNumber, F_OIDEQ,
+ ObjectIdGetDatum(relid));
+ ScanKeyInit(&key[2],
+ Anum_pg_depend_refobjsubid,
+ BTEqualStrategyNumber, F_INT4EQ,
+ Int32GetDatum((int32) attnum));
+
+ scan = systable_beginscan(depRel, DependReferenceIndexId, true,
+ NULL, 3, key);
+
+ while (HeapTupleIsValid(depTup = systable_getnext(scan)))
+ {
+ Form_pg_depend foundDep = (Form_pg_depend) GETSTRUCT(depTup);
+
+ if (foundDep->classid == AttrCompressionRelationId &&
+ foundDep->objid == acoid)
+ {
+ found = true;
+ break;
+ }
+ }
+ systable_endscan(scan);
+
+ if (!found)
+ recordDependencyOn(&acref, &attref, DEPENDENCY_INTERNAL);
+ }
+ heap_close(depRel, AccessShareLock);
+}
+
/*
* ALTER TABLE SET WITH OIDS
*
@@ -6671,6 +6870,10 @@ ATExecSetStorage(Relation rel, const char *colName, Node *newValue, LOCKMODE loc
errmsg("column data type %s can only have storage PLAIN",
format_type_be(attrtuple->atttypid))));
+ /* use default compression if storage is not PLAIN */
+ if (!OidIsValid(attrtuple->attcompression) && (newstorage != 'p'))
+ attrtuple->attcompression = DefaultCompressionOid;
+
CatalogTupleUpdate(attrelation, &tuple->t_self, tuple);
InvokeObjectPostAlterHook(RelationRelationId,
@@ -9572,6 +9775,12 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
colName)));
break;
+ case OCLASS_ATTR_COMPRESSION:
+
+ /* Just check that dependency is the right type */
+ Assert(foundDep->deptype == DEPENDENCY_INTERNAL);
+ break;
+
case OCLASS_DEFAULT:
/*
@@ -9672,7 +9881,8 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
if (!(foundDep->refclassid == TypeRelationId &&
foundDep->refobjid == attTup->atttypid) &&
!(foundDep->refclassid == CollationRelationId &&
- foundDep->refobjid == attTup->attcollation))
+ foundDep->refobjid == attTup->attcollation) &&
+ foundDep->refclassid != AttrCompressionRelationId)
elog(ERROR, "found unexpected dependency for column");
CatalogTupleDelete(depRel, &depTup->t_self);
@@ -9697,6 +9907,34 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
ReleaseSysCache(typeTuple);
+ if (rel->rd_rel->relkind == RELKIND_RELATION ||
+ rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE)
+ {
+ /* Setup attribute compression for new attribute */
+ if (OidIsValid(attTup->attcompression) && attTup->attstorage == 'p')
+ {
+ /* Set up rewrite of table */
+ attTup->attcompression = InvalidOid;
+ /* TODO: call the rewrite function here */
+ }
+ else if (OidIsValid(attTup->attcompression))
+ {
+ /* create new record */
+ ColumnCompression *compression = MakeColumnCompression(attTup->attcompression);
+
+ if (!IsBuiltinCompression(attTup->attcompression))
+ /* TODO: call the rewrite function here */;
+
+ attTup->attcompression = CreateAttributeCompression(attTup, compression, NULL, NULL);
+ }
+ else if (attTup->attstorage != 'p')
+ attTup->attcompression = DefaultCompressionOid;
+ else
+ attTup->attcompression = InvalidOid;
+ }
+ else
+ attTup->attcompression = InvalidOid;
+
CatalogTupleUpdate(attrelation, &heapTup->t_self, heapTup);
heap_close(attrelation, RowExclusiveLock);
@@ -9705,6 +9943,11 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel,
add_column_datatype_dependency(RelationGetRelid(rel), attnum, targettype);
add_column_collation_dependency(RelationGetRelid(rel), attnum, targetcollid);
+ /* Create dependency for new attribute compression */
+ if (OidIsValid(attTup->attcompression))
+ add_column_compression_dependency(RelationGetRelid(rel), attnum,
+ attTup->attcompression);
+
/*
* Drop any pg_statistic entry for the column, since it's now wrong type
*/
@@ -12733,6 +12976,93 @@ ATExecGenericOptions(Relation rel, List *options)
heap_freetuple(tuple);
}
+static ObjectAddress
+ATExecSetCompression(AlteredTableInfo *tab,
+ Relation rel,
+ const char *column,
+ ColumnCompression *compression,
+ LOCKMODE lockmode)
+{
+ Oid acoid;
+ Relation attrel;
+ HeapTuple atttuple;
+ Form_pg_attribute atttableform;
+ AttrNumber attnum;
+ bool need_rewrite;
+ Datum values[Natts_pg_attribute];
+ bool nulls[Natts_pg_attribute];
+ bool replace[Natts_pg_attribute];
+ List *preserved_amoids = NIL;
+ ObjectAddress address;
+
+ attrel = heap_open(AttributeRelationId, RowExclusiveLock);
+
+ atttuple = SearchSysCacheAttName(RelationGetRelid(rel), column);
+ if (!HeapTupleIsValid(atttuple))
+ ereport(ERROR,
+ (errcode(ERRCODE_UNDEFINED_COLUMN),
+ errmsg("column \"%s\" of relation \"%s\" does not exist",
+ column, RelationGetRelationName(rel))));
+
+ /* Prevent them from altering a system attribute */
+ atttableform = (Form_pg_attribute) GETSTRUCT(atttuple);
+ attnum = atttableform->attnum;
+ if (attnum <= 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot alter system column \"%s\"", column)));
+
+ /* Prevent from altering untoastable attributes with PLAIN storage */
+ if (atttableform->attstorage == 'p' && !TypeIsToastable(atttableform->atttypid))
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("column data type %s does not support compression",
+ format_type_be(atttableform->atttypid))));
+
+ /* Initialize buffers for new tuple values */
+ memset(values, 0, sizeof(values));
+ memset(nulls, false, sizeof(nulls));
+ memset(replace, false, sizeof(replace));
+
+ /* create pg_attr_compression record and its dependencies */
+ acoid = CreateAttributeCompression(atttableform, compression,
+ &need_rewrite, &preserved_amoids);
+ add_column_compression_dependency(atttableform->attrelid,
+ atttableform->attnum, acoid);
+
+ /*
+ * Save list of preserved access methods to cache which will be passed to
+ * toast_insert_or_update
+ */
+ if (need_rewrite)
+ /* TODO: set up rewrite rules here */;
+
+ atttableform->attcompression = acoid;
+ CatalogTupleUpdate(attrel, &atttuple->t_self, atttuple);
+
+ InvokeObjectPostAlterHook(RelationRelationId,
+ RelationGetRelid(rel),
+ atttableform->attnum);
+
+ ReleaseSysCache(atttuple);
+ heap_close(attrel, RowExclusiveLock);
+
+ /* make changes visible */
+ CommandCounterIncrement();
+
+ /*
+ * Normally cleanup is done in rewrite but in binary upgrade we should do
+ * it explicitly.
+ */
+ if (IsBinaryUpgrade)
+ CleanupAttributeCompression(RelationGetRelid(rel),
+ attnum, preserved_amoids);
+
+ ObjectAddressSet(address, AttrCompressionRelationId, acoid);
+ return address;
+}
+
+
/*
* Preparation phase for SET LOGGED/UNLOGGED
*
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 1c12075b01..3009d29cce 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2859,6 +2859,7 @@ _copyColumnDef(const ColumnDef *from)
COPY_STRING_FIELD(colname);
COPY_NODE_FIELD(typeName);
+ COPY_NODE_FIELD(compression);
COPY_SCALAR_FIELD(inhcount);
COPY_SCALAR_FIELD(is_local);
COPY_SCALAR_FIELD(is_not_null);
@@ -2878,6 +2879,18 @@ _copyColumnDef(const ColumnDef *from)
return newnode;
}
+static ColumnCompression *
+_copyColumnCompression(const ColumnCompression *from)
+{
+ ColumnCompression *newnode = makeNode(ColumnCompression);
+
+ COPY_STRING_FIELD(amname);
+ COPY_NODE_FIELD(options);
+ COPY_NODE_FIELD(preserve);
+
+ return newnode;
+}
+
static Constraint *
_copyConstraint(const Constraint *from)
{
@@ -5553,6 +5566,9 @@ copyObjectImpl(const void *from)
case T_ColumnDef:
retval = _copyColumnDef(from);
break;
+ case T_ColumnCompression:
+ retval = _copyColumnCompression(from);
+ break;
case T_Constraint:
retval = _copyConstraint(from);
break;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 6a971d0141..06b34a032c 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2549,6 +2549,7 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b)
{
COMPARE_STRING_FIELD(colname);
COMPARE_NODE_FIELD(typeName);
+ COMPARE_NODE_FIELD(compression);
COMPARE_SCALAR_FIELD(inhcount);
COMPARE_SCALAR_FIELD(is_local);
COMPARE_SCALAR_FIELD(is_not_null);
@@ -2568,6 +2569,16 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b)
return true;
}
+static bool
+_equalColumnCompression(const ColumnCompression *a, const ColumnCompression *b)
+{
+ COMPARE_STRING_FIELD(amname);
+ COMPARE_NODE_FIELD(options);
+ COMPARE_NODE_FIELD(preserve);
+
+ return true;
+}
+
static bool
_equalConstraint(const Constraint *a, const Constraint *b)
{
@@ -3627,6 +3638,9 @@ equal(const void *a, const void *b)
case T_ColumnDef:
retval = _equalColumnDef(a, b);
break;
+ case T_ColumnCompression:
+ retval = _equalColumnCompression(a, b);
+ break;
case T_Constraint:
retval = _equalConstraint(a, b);
break;
diff --git a/src/backend/nodes/nodeFuncs.c b/src/backend/nodes/nodeFuncs.c
index a10014f755..07ec871980 100644
--- a/src/backend/nodes/nodeFuncs.c
+++ b/src/backend/nodes/nodeFuncs.c
@@ -3648,6 +3648,8 @@ raw_expression_tree_walker(Node *node,
if (walker(coldef->typeName, context))
return true;
+ if (walker(coldef->compression, context))
+ return true;
if (walker(coldef->raw_default, context))
return true;
if (walker(coldef->collClause, context))
@@ -3655,6 +3657,14 @@ raw_expression_tree_walker(Node *node,
/* for now, constraints are ignored */
}
break;
+ case T_ColumnCompression:
+ {
+ ColumnCompression *colcmp = (ColumnCompression *) node;
+
+ if (walker(colcmp->options, context))
+ return true;
+ }
+ break;
case T_IndexElem:
{
IndexElem *indelem = (IndexElem *) node;
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 979d523e00..06097965c7 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -2859,6 +2859,7 @@ _outColumnDef(StringInfo str, const ColumnDef *node)
WRITE_STRING_FIELD(colname);
WRITE_NODE_FIELD(typeName);
+ WRITE_NODE_FIELD(compression);
WRITE_INT_FIELD(inhcount);
WRITE_BOOL_FIELD(is_local);
WRITE_BOOL_FIELD(is_not_null);
@@ -2876,6 +2877,16 @@ _outColumnDef(StringInfo str, const ColumnDef *node)
WRITE_LOCATION_FIELD(location);
}
+static void
+_outColumnCompression(StringInfo str, const ColumnCompression *node)
+{
+ WRITE_NODE_TYPE("COLUMNCOMPRESSION");
+
+ WRITE_STRING_FIELD(amname);
+ WRITE_NODE_FIELD(options);
+ WRITE_NODE_FIELD(preserve);
+}
+
static void
_outTypeName(StringInfo str, const TypeName *node)
{
@@ -4181,6 +4192,9 @@ outNode(StringInfo str, const void *obj)
case T_ColumnDef:
_outColumnDef(str, obj);
break;
+ case T_ColumnCompression:
+ _outColumnCompression(str, obj);
+ break;
case T_TypeName:
_outTypeName(str, obj);
break;
diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c
index 17b54b20cc..2b909fc949 100644
--- a/src/backend/parser/parse_utilcmd.c
+++ b/src/backend/parser/parse_utilcmd.c
@@ -1061,6 +1061,13 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla
else
def->storage = 0;
+ /* Likewise, copy compression if requested */
+ if (table_like_clause->options & CREATE_TABLE_LIKE_COMPRESSION
+ && OidIsValid(attribute->attcompression))
+ def->compression = MakeColumnCompression(attribute->attcompression);
+ else
+ def->compression = NULL;
+
/* Likewise, copy comment if requested */
if ((table_like_clause->options & CREATE_TABLE_LIKE_COMMENTS) &&
(comment = GetComment(attribute->attrelid,
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 5792cd14a0..300eb6fd4c 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -3038,7 +3038,7 @@ ReorderBufferToastReplace(ReorderBuffer *rb, ReorderBufferTXN *txn,
VARSIZE(chunk) - VARHDRSZ);
data_done += VARSIZE(chunk) - VARHDRSZ;
}
- Assert(data_done == toast_pointer.va_extsize);
+ Assert(data_done == VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer));
/* make sure its marked as compressed or not */
if (VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer))
diff --git a/src/backend/utils/adt/pseudotypes.c b/src/backend/utils/adt/pseudotypes.c
index dbe67cdb4c..4a97099cd0 100644
--- a/src/backend/utils/adt/pseudotypes.c
+++ b/src/backend/utils/adt/pseudotypes.c
@@ -418,3 +418,4 @@ PSEUDOTYPE_DUMMY_IO_FUNCS(internal);
PSEUDOTYPE_DUMMY_IO_FUNCS(opaque);
PSEUDOTYPE_DUMMY_IO_FUNCS(anyelement);
PSEUDOTYPE_DUMMY_IO_FUNCS(anynonarray);
+PSEUDOTYPE_DUMMY_IO_FUNCS(compression_am_handler);
diff --git a/src/backend/utils/cache/syscache.c b/src/backend/utils/cache/syscache.c
index 2b381782a3..0c97851cbd 100644
--- a/src/backend/utils/cache/syscache.c
+++ b/src/backend/utils/cache/syscache.c
@@ -27,6 +27,7 @@
#include "catalog/pg_am.h"
#include "catalog/pg_amop.h"
#include "catalog/pg_amproc.h"
+#include "catalog/pg_attr_compression.h"
#include "catalog/pg_auth_members.h"
#include "catalog/pg_authid.h"
#include "catalog/pg_cast.h"
@@ -187,6 +188,17 @@ static const struct cachedesc cacheinfo[] = {
},
16
},
+ {AttrCompressionRelationId, /* ATTCOMPRESSIONOID */
+ AttrCompressionIndexId,
+ 1,
+ {
+ Anum_pg_attr_compression_acoid,
+ 0,
+ 0,
+ 0
+ },
+ 8,
+ },
{AttributeRelationId, /* ATTNAME */
AttributeRelidNameIndexId,
2,
diff --git a/src/include/access/cmapi.h b/src/include/access/cmapi.h
new file mode 100644
index 0000000000..9e48f0d49f
--- /dev/null
+++ b/src/include/access/cmapi.h
@@ -0,0 +1,80 @@
+/*-------------------------------------------------------------------------
+ *
+ * cmapi.h
+ * API for Postgres compression AM.
+ *
+ * Copyright (c) 2015-2017, PostgreSQL Global Development Group
+ *
+ * src/include/access/cmapi.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef CMAPI_H
+#define CMAPI_H
+
+#include "postgres.h"
+#include "catalog/pg_attr_compression.h"
+#include "catalog/pg_attribute.h"
+#include "nodes/pg_list.h"
+
+#define IsBuiltinCompression(cmid) ((cmid) < FirstBootstrapObjectId)
+#define DefaultCompressionOid (InvalidOid)
+
+typedef struct CompressionAmRoutine CompressionAmRoutine;
+
+/*
+ * CompressionAmOptions contains all information needed to compress varlena.
+ *
+ * For optimization purposes it will be created once for each attribute
+ * compression and stored in cache, until its renewal on global cache reset,
+ * or until deletion of related attribute compression.
+ */
+typedef struct CompressionAmOptions
+{
+ Oid acoid; /* Oid of attribute compression,
+ should go always first */
+ Oid amoid; /* Oid of compression access method */
+ List *acoptions; /* Parsed options, used for comparison */
+ CompressionAmRoutine *amroutine; /* compression access method routine */
+ MemoryContext mcxt;
+
+ /* result of cminitstate function will be put here */
+ void *acstate;
+} CompressionAmOptions;
+
+typedef void (*cmcheck_function) (Form_pg_attribute att, List *options);
+typedef struct varlena *(*cmcompress_function)
+ (CompressionAmOptions *cmoptions, const struct varlena *value);
+typedef void *(*cminitstate_function) (Oid acoid, List *options);
+
+/*
+ * API struct for a compression AM.
+ *
+ * 'cmcheck' - called when attribute is linking with compression method.
+ * This function should check compability of compression method with
+ * the attribute and its options.
+ *
+ * 'cminitstate' - called when CompressionAmOptions instance is created.
+ * Should return pointer to a memory in a caller memory context, or NULL.
+ * Could be used to pass some internal state between compression function
+ * calls, like internal structure for parsed compression options.
+ *
+ * 'cmcompress' and 'cmdecompress' - varlena compression functions.
+ */
+struct CompressionAmRoutine
+{
+ NodeTag type;
+
+ cmcheck_function cmcheck; /* can be NULL */
+ cminitstate_function cminitstate; /* can be NULL */
+ cmcompress_function cmcompress;
+ cmcompress_function cmdecompress;
+};
+
+/* access/compression/cmapi.c */
+extern CompressionAmRoutine *InvokeCompressionAmHandler(Oid amhandler);
+extern List *GetAttrCompressionOptions(Oid acoid);
+extern Oid GetAttrCompressionAmOid(Oid acoid);
+
+#endif /* CMAPI_H */
diff --git a/src/include/access/reloptions.h b/src/include/access/reloptions.h
index 4022c14a83..c9302347e8 100644
--- a/src/include/access/reloptions.h
+++ b/src/include/access/reloptions.h
@@ -259,7 +259,6 @@ extern void add_string_reloption(bits32 kinds, const char *name, const char *des
extern Datum transformRelOptions(Datum oldOptions, List *defList,
const char *namspace, char *validnsps[],
bool ignoreOids, bool isReset);
-extern List *untransformRelOptions(Datum options);
extern bytea *extractRelOptions(HeapTuple tuple, TupleDesc tupdesc,
amoptions_function amoptions);
extern relopt_value *parseRelOptions(Datum options, bool validate,
@@ -270,6 +269,8 @@ extern void fillRelOptions(void *rdopts, Size basesize,
relopt_value *options, int numoptions,
bool validate,
const relopt_parse_elt *elems, int nelems);
+extern char *formatRelOptions(List *options);
+extern List *untransformRelOptions(Datum options);
extern bytea *default_reloptions(Datum reloptions, bool validate,
relopt_kind kind);
diff --git a/src/include/access/tuptoaster.h b/src/include/access/tuptoaster.h
index f99291e30d..793367ff58 100644
--- a/src/include/access/tuptoaster.h
+++ b/src/include/access/tuptoaster.h
@@ -13,9 +13,11 @@
#ifndef TUPTOASTER_H
#define TUPTOASTER_H
+#include "access/cmapi.h"
#include "access/htup_details.h"
#include "storage/lockdefs.h"
#include "utils/relcache.h"
+#include "utils/hsearch.h"
/*
* This enables de-toasting of index entries. Needed until VACUUM is
@@ -101,6 +103,15 @@
/* Size of an EXTERNAL datum that contains an indirection pointer */
#define INDIRECT_POINTER_SIZE (VARHDRSZ_EXTERNAL + sizeof(varatt_indirect))
+/*
+ * va_extinfo in varatt_external contains actual length of the external data
+ * and optional flags
+ */
+#define VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer) \
+ ((toast_pointer).va_extinfo & 0x3FFFFFFF)
+#define VARATT_EXTERNAL_IS_CUSTOM_COMPRESSED(toast_pointer) \
+ (((toast_pointer).va_extinfo >> 30) == 0x02)
+
/*
* Testing whether an externally-stored value is compressed now requires
* comparing extsize (the actual length of the external data) to rawsize
@@ -109,7 +120,7 @@
* saves space, so we expect either equality or less-than.
*/
#define VARATT_EXTERNAL_IS_COMPRESSED(toast_pointer) \
- ((toast_pointer).va_extsize < (toast_pointer).va_rawsize - VARHDRSZ)
+ (VARATT_EXTERNAL_GET_EXTSIZE(toast_pointer) < (toast_pointer).va_rawsize - VARHDRSZ)
/*
* Macro to fetch the possibly-unaligned contents of an EXTERNAL datum
@@ -126,6 +137,16 @@ do { \
memcpy(&(toast_pointer), VARDATA_EXTERNAL(attre), sizeof(toast_pointer)); \
} while (0)
+/*
+ * Used to pass to preserved compression access methods from
+ * ALTER TABLE SET COMPRESSION into toast_insert_or_update.
+ */
+typedef struct AttrCmPreservedInfo
+{
+ AttrNumber attnum;
+ List *preserved_amoids;
+} AttrCmPreservedInfo;
+
/* ----------
* toast_insert_or_update -
*
@@ -134,7 +155,7 @@ do { \
*/
extern HeapTuple toast_insert_or_update(Relation rel,
HeapTuple newtup, HeapTuple oldtup,
- int options);
+ int options, HTAB *preserved_cmlist);
/* ----------
* toast_delete -
@@ -210,7 +231,7 @@ extern HeapTuple toast_build_flattened_tuple(TupleDesc tupleDesc,
* Create a compressed version of a varlena datum, if possible
* ----------
*/
-extern Datum toast_compress_datum(Datum value);
+extern Datum toast_compress_datum(Datum value, Oid cmoptoid);
/* ----------
* toast_raw_datum_size -
@@ -236,4 +257,19 @@ extern Size toast_datum_size(Datum value);
*/
extern Oid toast_get_valid_index(Oid toastoid, LOCKMODE lock);
+/*
+ * lookup_compression_am_options -
+ *
+ * Return cached CompressionAmOptions for specified attribute compression.
+ */
+extern CompressionAmOptions *lookup_compression_am_options(Oid acoid);
+
+/*
+ * toast_set_compressed_datum_info -
+ *
+ * Save metadata in compressed datum
+ */
+extern void toast_set_compressed_datum_info(struct varlena *val, Oid cmid,
+ int32 rawsize);
+
#endif /* TUPTOASTER_H */
diff --git a/src/include/catalog/dependency.h b/src/include/catalog/dependency.h
index 46c271a46c..cb19cb3e41 100644
--- a/src/include/catalog/dependency.h
+++ b/src/include/catalog/dependency.h
@@ -180,10 +180,11 @@ typedef enum ObjectClass
OCLASS_PUBLICATION, /* pg_publication */
OCLASS_PUBLICATION_REL, /* pg_publication_rel */
OCLASS_SUBSCRIPTION, /* pg_subscription */
- OCLASS_TRANSFORM /* pg_transform */
+ OCLASS_TRANSFORM, /* pg_transform */
+ OCLASS_ATTR_COMPRESSION /* pg_attr_compression */
} ObjectClass;
-#define LAST_OCLASS OCLASS_TRANSFORM
+#define LAST_OCLASS OCLASS_ATTR_COMPRESSION
/* flag bits for performDeletion/performMultipleDeletions: */
#define PERFORM_DELETION_INTERNAL 0x0001 /* internal action */
diff --git a/src/include/catalog/indexing.h b/src/include/catalog/indexing.h
index 24915824ca..dc41e50bd8 100644
--- a/src/include/catalog/indexing.h
+++ b/src/include/catalog/indexing.h
@@ -87,6 +87,11 @@ DECLARE_UNIQUE_INDEX(pg_attrdef_adrelid_adnum_index, 2656, on pg_attrdef using b
DECLARE_UNIQUE_INDEX(pg_attrdef_oid_index, 2657, on pg_attrdef using btree(oid oid_ops));
#define AttrDefaultOidIndexId 2657
+DECLARE_UNIQUE_INDEX(pg_attr_compression_index, 4003, on pg_attr_compression using btree(acoid oid_ops));
+#define AttrCompressionIndexId 4003
+DECLARE_INDEX(pg_attr_compression_relid_attnum_index, 4004, on pg_attr_compression using btree(acrelid oid_ops, acattnum int2_ops));
+#define AttrCompressionRelidAttnumIndexId 4004
+
DECLARE_UNIQUE_INDEX(pg_attribute_relid_attnam_index, 2658, on pg_attribute using btree(attrelid oid_ops, attname name_ops));
#define AttributeRelidNameIndexId 2658
DECLARE_UNIQUE_INDEX(pg_attribute_relid_attnum_index, 2659, on pg_attribute using btree(attrelid oid_ops, attnum int2_ops));
diff --git a/src/include/catalog/pg_attr_compression.dat b/src/include/catalog/pg_attr_compression.dat
new file mode 100644
index 0000000000..30faae0de4
--- /dev/null
+++ b/src/include/catalog/pg_attr_compression.dat
@@ -0,0 +1,23 @@
+#----------------------------------------------------------------------
+#
+# pg_attr_compression.dat
+# Initial contents of the pg_attr_compression system relation.
+#
+# Predefined compression options for builtin compression access methods.
+# It is safe to use Oids that equal to Oids of access methods, since
+# these are just numbers and not system Oids.
+#
+# Note that predefined options should have 0 in acrelid and acattnum.
+#
+# Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+# Portions Copyright (c) 1994, Regents of the University of California
+#
+# src/include/catalog/pg_attr_compression.dat
+#
+#----------------------------------------------------------------------
+
+[
+
+
+
+]
diff --git a/src/include/catalog/pg_attr_compression.h b/src/include/catalog/pg_attr_compression.h
new file mode 100644
index 0000000000..563aa65a2a
--- /dev/null
+++ b/src/include/catalog/pg_attr_compression.h
@@ -0,0 +1,53 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_attr_compression.h
+ *
+ * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/catalog/pg_attr_compression.h
+ *
+ * NOTES
+ * The Catalog.pm module reads this file and derives schema
+ * information.
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_ATTR_COMPRESSION_H
+#define PG_ATTR_COMPRESSION_H
+
+#include "catalog/genbki.h"
+#include "catalog/pg_am_d.h"
+#include "catalog/pg_attr_compression_d.h"
+
+/* ----------------
+ * pg_attr_compression definition. cpp turns this into
+ * typedef struct FormData_pg_attr_compression
+ * ----------------
+ */
+CATALOG(pg_attr_compression,4001,AttrCompressionRelationId) BKI_WITHOUT_OIDS
+{
+ Oid acoid; /* attribute compression oid */
+ NameData acname; /* name of compression AM */
+ Oid acrelid BKI_DEFAULT(0); /* attribute relation */
+ int16 acattnum BKI_DEFAULT(0); /* attribute number in the relation */
+
+#ifdef CATALOG_VARLEN /* variable-length fields start here */
+ text acoptions[1] BKI_DEFAULT(_null_); /* specific options from WITH */
+#endif
+} FormData_pg_attr_compression;
+
+/* ----------------
+ * Form_pg_attr_compresssion corresponds to a pointer to a tuple with
+ * the format of pg_attr_compression relation.
+ * ----------------
+ */
+typedef FormData_pg_attr_compression *Form_pg_attr_compression;
+
+#ifdef EXPOSE_TO_CLIENT_CODE
+/* builtin attribute compression Oids */
+#define PGLZ_AC_OID (PGLZ_COMPRESSION_AM_OID)
+#define ZLIB_AC_OID (ZLIB_COMPRESSION_AM_OID)
+#endif
+
+#endif /* PG_ATTR_COMPRESSION_H */
diff --git a/src/include/catalog/pg_attribute.h b/src/include/catalog/pg_attribute.h
index dc36753ede..08e074ea4d 100644
--- a/src/include/catalog/pg_attribute.h
+++ b/src/include/catalog/pg_attribute.h
@@ -160,6 +160,9 @@ CATALOG(pg_attribute,1249,AttributeRelationId) BKI_BOOTSTRAP BKI_WITHOUT_OIDS BK
/* attribute's collation */
Oid attcollation;
+ /* attribute's compression options or InvalidOid */
+ Oid attcompression BKI_DEFAULT(0);
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* NOTE: The following fields are not present in tuple descriptors. */
@@ -184,10 +187,10 @@ CATALOG(pg_attribute,1249,AttributeRelationId) BKI_BOOTSTRAP BKI_WITHOUT_OIDS BK
* ATTRIBUTE_FIXED_PART_SIZE is the size of the fixed-layout,
* guaranteed-not-null part of a pg_attribute row. This is in fact as much
* of the row as gets copied into tuple descriptors, so don't expect you
- * can access fields beyond attcollation except in a real tuple!
+ * can access fields beyond attcompression except in a real tuple!
*/
#define ATTRIBUTE_FIXED_PART_SIZE \
- (offsetof(FormData_pg_attribute,attcollation) + sizeof(Oid))
+ (offsetof(FormData_pg_attribute,attcompression) + sizeof(Oid))
/* ----------------
* Form_pg_attribute corresponds to a pointer to a tuple with
diff --git a/src/include/catalog/pg_class.dat b/src/include/catalog/pg_class.dat
index 9fffdef379..49643b934e 100644
--- a/src/include/catalog/pg_class.dat
+++ b/src/include/catalog/pg_class.dat
@@ -36,7 +36,7 @@
reloftype => '0', relowner => 'PGUID', relam => '0', relfilenode => '0',
reltablespace => '0', relpages => '0', reltuples => '0', relallvisible => '0',
reltoastrelid => '0', relhasindex => 'f', relisshared => 'f',
- relpersistence => 'p', relkind => 'r', relnatts => '24', relchecks => '0',
+ relpersistence => 'p', relkind => 'r', relnatts => '25', relchecks => '0',
relhasoids => 'f', relhasrules => 'f', relhastriggers => 'f',
relhassubclass => 'f', relrowsecurity => 'f', relforcerowsecurity => 'f',
relispopulated => 't', relreplident => 'n', relispartition => 'f',
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 40d54ed030..3242179a8a 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -6915,6 +6915,9 @@
descr => 'bytes required to store the value, perhaps with compression',
proname => 'pg_column_size', provolatile => 's', prorettype => 'int4',
proargtypes => 'any', prosrc => 'pg_column_size' },
+{ oid => '4008', descr => 'list of compression methods used by the column',
+ proname => 'pg_column_compression', provolatile => 's', prorettype => 'text',
+ proargtypes => 'regclass text', prosrc => 'pg_column_compression' },
{ oid => '2322',
descr => 'total disk space usage for the specified tablespace',
proname => 'pg_tablespace_size', provolatile => 'v', prorettype => 'int8',
@@ -7084,6 +7087,13 @@
{ oid => '3312', descr => 'I/O',
proname => 'tsm_handler_out', prorettype => 'cstring',
proargtypes => 'tsm_handler', prosrc => 'tsm_handler_out' },
+{ oid => '4006', descr => 'I/O',
+ proname => 'compression_am_handler_in', proisstrict => 'f',
+ prorettype => 'compression_am_handler', proargtypes => 'cstring',
+ prosrc => 'compression_am_handler_in' },
+{ oid => '4007', descr => 'I/O',
+ proname => 'compression_am_handler_out', prorettype => 'cstring',
+ proargtypes => 'compression_am_handler', prosrc => 'compression_am_handler_out' },
# tablesample method handlers
{ oid => '3313', descr => 'BERNOULLI tablesample method handler',
diff --git a/src/include/catalog/pg_type.dat b/src/include/catalog/pg_type.dat
index 48e01cd694..722b11674f 100644
--- a/src/include/catalog/pg_type.dat
+++ b/src/include/catalog/pg_type.dat
@@ -929,6 +929,11 @@
typcategory => 'P', typinput => 'index_am_handler_in',
typoutput => 'index_am_handler_out', typreceive => '-', typsend => '-',
typalign => 'i' },
+{ oid => '4005',
+ typname => 'compression_am_handler', typlen => '4', typbyval => 't', typtype => 'p',
+ typcategory => 'P', typinput => 'compression_am_handler_in',
+ typoutput => 'compression_am_handler_out', typreceive => '-', typsend => '-',
+ typalign => 'i' },
{ oid => '3310',
typname => 'tsm_handler', typlen => '4', typbyval => 't', typtype => 'p',
typcategory => 'P', typinput => 'tsm_handler_in',
diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h
index 8fc9e424cf..9acbebb0bc 100644
--- a/src/include/commands/defrem.h
+++ b/src/include/commands/defrem.h
@@ -145,6 +145,7 @@ extern Oid RemoveUserMapping(DropUserMappingStmt *stmt);
extern void RemoveUserMappingById(Oid umId);
extern void CreateForeignTable(CreateForeignTableStmt *stmt, Oid relid);
extern void ImportForeignSchema(ImportForeignSchemaStmt *stmt);
+extern Datum optionListToArray(List *options, bool sorted);
extern Datum transformGenericOptions(Oid catalogId,
Datum oldOptions,
List *options,
@@ -154,8 +155,21 @@ extern Datum transformGenericOptions(Oid catalogId,
extern ObjectAddress CreateAccessMethod(CreateAmStmt *stmt);
extern void RemoveAccessMethodById(Oid amOid);
extern Oid get_index_am_oid(const char *amname, bool missing_ok);
+extern Oid get_compression_am_oid(const char *amname, bool missing_ok);
extern Oid get_am_oid(const char *amname, bool missing_ok);
extern char *get_am_name(Oid amOid);
+extern regproc get_am_handler_oid(Oid amOid, char amtype, bool noerror);
+
+/* commands/compressioncmds.c */
+extern ColumnCompression *MakeColumnCompression(Oid acoid);
+extern Oid CreateAttributeCompression(Form_pg_attribute attr,
+ ColumnCompression *compression,
+ bool *need_rewrite,
+ List **preserved_amoids);
+extern void RemoveAttributeCompression(Oid acoid);
+extern void CheckCompressionMismatch(ColumnCompression *c1,
+ ColumnCompression *c2, const char *attributeName);
+void CleanupAttributeCompression(Oid relid, AttrNumber attnum, List *keepAmOids);
/* support routines in commands/define.c */
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 03cee8ec88..c54dfc7462 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -505,7 +505,8 @@ typedef enum NodeTag
T_IndexAmRoutine, /* in access/amapi.h */
T_TsmRoutine, /* in access/tsmapi.h */
T_ForeignKeyCacheInfo, /* in utils/rel.h */
- T_CallContext /* in nodes/parsenodes.h */
+ T_CallContext, /* in nodes/parsenodes.h */
+ T_CompressionAmRoutine /* in access/cmapi.h */
} NodeTag;
/*
diff --git a/src/include/postgres.h b/src/include/postgres.h
index b596fcb513..916c75c7d8 100644
--- a/src/include/postgres.h
+++ b/src/include/postgres.h
@@ -55,7 +55,7 @@
/*
* struct varatt_external is a traditional "TOAST pointer", that is, the
* information needed to fetch a Datum stored out-of-line in a TOAST table.
- * The data is compressed if and only if va_extsize < va_rawsize - VARHDRSZ.
+ * The data is compressed if and only if size in va_extinfo < va_rawsize - VARHDRSZ.
* This struct must not contain any padding, because we sometimes compare
* these pointers using memcmp.
*
@@ -67,7 +67,8 @@
typedef struct varatt_external
{
int32 va_rawsize; /* Original data size (includes header) */
- int32 va_extsize; /* External saved size (doesn't) */
+ uint32 va_extinfo; /* External saved size (without header) and
+ * flags */
Oid va_valueid; /* Unique ID of value within TOAST table */
Oid va_toastrelid; /* RelID of TOAST table containing it */
} varatt_external;
@@ -145,9 +146,18 @@ typedef union
struct /* Compressed-in-line format */
{
uint32 va_header;
- uint32 va_rawsize; /* Original data size (excludes header) */
+ uint32 va_info; /* Original data size (excludes header) and
+ * flags */
char va_data[FLEXIBLE_ARRAY_MEMBER]; /* Compressed data */
} va_compressed;
+ struct /* Compressed-in-line format */
+ {
+ uint32 va_header;
+ uint32 va_info; /* Original data size (excludes header) and
+ * flags */
+ Oid va_cmid; /* Oid of compression options */
+ char va_data[FLEXIBLE_ARRAY_MEMBER]; /* Compressed data */
+ } va_custom_compressed;
} varattrib_4b;
typedef struct
@@ -280,8 +290,14 @@ typedef struct
#define VARDATA_1B(PTR) (((varattrib_1b *) (PTR))->va_data)
#define VARDATA_1B_E(PTR) (((varattrib_1b_e *) (PTR))->va_data)
+/* va_info in va_compress contains raw size of datum and optional flags */
#define VARRAWSIZE_4B_C(PTR) \
- (((varattrib_4b *) (PTR))->va_compressed.va_rawsize)
+ (((varattrib_4b *) (PTR))->va_compressed.va_info & 0x3FFFFFFF)
+#define VARFLAGS_4B_C(PTR) \
+ (((varattrib_4b *) (PTR))->va_compressed.va_info >> 30)
+
+#define VARHDRSZ_CUSTOM_COMPRESSED \
+ (offsetof(varattrib_4b, va_custom_compressed.va_data))
/* Externally visible macros */
@@ -310,6 +326,8 @@ typedef struct
#define VARDATA_EXTERNAL(PTR) VARDATA_1B_E(PTR)
#define VARATT_IS_COMPRESSED(PTR) VARATT_IS_4B_C(PTR)
+#define VARATT_IS_CUSTOM_COMPRESSED(PTR) (VARATT_IS_4B_C(PTR) && \
+ (VARFLAGS_4B_C(PTR) == 0x02))
#define VARATT_IS_EXTERNAL(PTR) VARATT_IS_1B_E(PTR)
#define VARATT_IS_EXTERNAL_ONDISK(PTR) \
(VARATT_IS_EXTERNAL(PTR) && VARTAG_EXTERNAL(PTR) == VARTAG_ONDISK)
diff --git a/src/include/utils/syscache.h b/src/include/utils/syscache.h
index 4f333586ee..48051b8294 100644
--- a/src/include/utils/syscache.h
+++ b/src/include/utils/syscache.h
@@ -37,6 +37,7 @@ enum SysCacheIdentifier
AMOPOPID,
AMOPSTRATEGY,
AMPROCNUM,
+ ATTCOMPRESSIONOID,
ATTNAME,
ATTNUM,
AUTHMEMMEMROLE,
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 03867cbce5..2310208331 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -350,6 +350,7 @@ CollectedCommand
CollectedCommandType
ColorTrgm
ColorTrgmInfo
+ColumnCompression
ColumnCompareData
ColumnDef
ColumnIOData
@@ -375,6 +376,8 @@ CompositeIOData
CompositeTypeStmt
CompoundAffixFlag
CompressionAlgorithm
+CompressionAmOptions
+CompressionAmRoutine
CompressorState
ConditionVariable
ConditionalStack
--
2.18.0
--MP_/w9zXrmKKcGlmI.IAcoy33yI
Content-Type: text/x-patch
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
filename=0003-Add-rewrite-rules-and-tupdesc-flags-v17.patch
view thread (8+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH 2/8] Add compression catalog tables and the basic infrastructure
In-Reply-To: <no-message-id-235043@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox