From: Ildus Kurbangaliev Date: Mon, 18 Jun 2018 15:26:13 +0300 Subject: [PATCH 2/8] Add compression catalog tables and the basic infrastructure --- 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 #include +#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