Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtp (Exim 4.84_2) (envelope-from ) id 1dpztK-0005AW-Nc for pgsql-hackers@arkaria.postgresql.org; Thu, 07 Sep 2017 16:42:59 +0000 Received: from localhost ([127.0.0.1] helo=postgresql.org) by malur.postgresql.org with smtp (Exim 4.84_2) (envelope-from ) id 1dpztK-0001dy-5j for pgsql-hackers@arkaria.postgresql.org; Thu, 07 Sep 2017 16:42:58 +0000 Received: from magus.postgresql.org ([2a02:c0:301:0:ffff::29]) by malur.postgresql.org with esmtps (TLS1.2:ECDHE_RSA_AES_256_CBC_SHA384:256) (Exim 4.84_2) (envelope-from ) id 1dpztH-0001cT-MS for pgsql-hackers@postgresql.org; Thu, 07 Sep 2017 16:42:55 +0000 Received: from mail.postgrespro.ru ([93.174.131.138]) by magus.postgresql.org with esmtp (Exim 4.84_2) (envelope-from ) id 1dpzsz-0002jc-Sh for pgsql-hackers@postgresql.org; Thu, 07 Sep 2017 16:42:55 +0000 Received: from localhost (localhost [127.0.0.1]) by mail.postgrespro.ru (Postfix) with ESMTP id 391EA21C1D1D for ; Thu, 7 Sep 2017 19:42:37 +0300 (MSK) X-Virus-Scanned: Debian amavisd-new at postgrespro.ru Received: from wp.localdomain (unknown [192.168.27.1]) (using TLSv1.2 with cipher ECDHE-RSA-AES256-GCM-SHA384 (256/256 bits)) (Client did not present a certificate) by mail.postgrespro.ru (Postfix) with ESMTPSA id BFE4021C1D04 for ; Thu, 7 Sep 2017 19:42:36 +0300 (MSK) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/simple; d=postgrespro.ru; s=mail; t=1504802556; bh=aX4w1zqSycYYI7z5A0U9YaGsANec3JK9A5OhA+I874A=; h=Date:From:To:Subject; b=jDgCDKvMczP8dgidhl8Y4mSpfn8w0WnZpBhvj4P4qwD5LNvVF3Wry8eHeFeCxmu+k Rmo5DyAyY4nRsx6a0sVlTkRlXvfpYGnOZUb6mad3zueILismnx4x9IVn4iAV92llIa CKQWSlT+yYnSNN/iULq8h3QA0UyhhmJpG1C+nq7Y= Date: Thu, 7 Sep 2017 19:42:36 +0300 From: Ildus Kurbangaliev To: pgsql-hackers@postgresql.org Subject: Custom compression methods Message-ID: <20170907194236.4cefce96@wp.localdomain> Organization: Postgres Professional X-Mailer: Claws Mail 3.15.0-dirty (GTK+ 2.24.31; x86_64-pc-linux-gnu) MIME-Version: 1.0 Content-Type: multipart/mixed; boundary="MP_/hLqNptav_zmIGzXV5rPJ9_y" List-Archive: List-Help: List-ID: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: X-Mailing-List: pgsql-hackers Precedence: bulk Sender: pgsql-hackers-owner@postgresql.org --MP_/hLqNptav_zmIGzXV5rPJ9_y Content-Type: text/plain; charset=US-ASCII Content-Transfer-Encoding: 7bit Content-Disposition: inline Hello hackers! I've attached a patch that implements custom compression methods. This patch is based on Nikita Glukhov's code (which he hasn't publish in mailing lists) for jsonb compression. This is early but working version of the patch, and there are still few fixes and features that should be implemented (like pg_dump support and support of compression options for types), and it requires more testing. But I'd like to get some feedback at the current stage first. There's been a proposal [1] of Alexander Korotkov and some discussion about custom compression methods before. This is an implementation of per-datum compression. Syntax is similar to the one in proposal but not the same. Syntax: CREATE COMPRESSION METHOD HANDLER ; DROP COMPRESSION METHOD ; Compression handler is a function that returns a structure containing compression routines: - configure - function called when the compression method applied to an attribute - drop - called when the compression method is removed from an attribute - compress - compress function - decompress - decompress function User can create compressed columns with the commands below: CREATE TABLE t(a tsvector COMPRESSED WITH ); ALTER TABLE t ALTER COLUMN a SET COMPRESSED WITH ; ALTER TABLE t ALTER COLUMN a SET NOT COMPRESSED; Also there is syntax of binding compression methods to types: ALTER TYPE SET COMPRESSED ; ALTER TYPE SET NOT COMPRESSED; There are two new tables in the catalog, pg_compression and pg_compression_opt. pg_compression is used as storage of compression methods, and pg_compression_opt is used to store specific compression options for particular column. When user binds a compression method to some column a new record in pg_compression_opt is created and all further attribute values will contain compression options Oid while old values will remain unchanged. And when we alter a compression method for the attribute it won't change previous record in pg_compression_opt. Instead it'll create a new one and new values will be stored with new Oid. That way there is no need of recompression of the old tuples. And also tuples containing compressed datums can be copied to other tables so records in pg_compression_opt shouldn't be removed. In the current patch they can be removed with DROP COMPRESSION METHOD CASCADE, but after that decompression won't be possible on compressed tuples. Maybe CASCADE should keep compression options. I haven't changed the base logic of working with compressed datums. It means that custom compressed datums behave exactly the same as current LZ compressed datums, and the logic differs only in toast_compress_datum and toast_decompress_datum. This patch doesn't break backward compability and should work seamlessly with older version of database. I used one of two free bits in `va_rawsize` from `varattrib_4b->va_compressed` as flag of custom compressed datums. Also I renamed it to `va_info` since it contains not only rawsize now. The patch also includes custom compression method for tsvector which is used in tests. [1] https://www.postgresql.org/message-id/CAPpHfdsdTA5uZeq6MNXL5ZRuNx%2BSig4ykWzWEAfkC6ZKMDy6%3DQ%40mail.gmail.com -- --- Ildus Kurbangaliev Postgres Professional: http://www.postgrespro.com Russian Postgres Company --MP_/hLqNptav_zmIGzXV5rPJ9_y Content-Type: text/x-patch Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename=custom_compression_methods_v1.patch diff --git a/contrib/test_decoding/expected/ddl.out b/contrib/test_decoding/expected/ddl.out index 1e22c1eefc..766ced401f 100644 --- a/contrib/test_decoding/expected/ddl.out +++ b/contrib/test_decoding/expected/ddl.out @@ -416,12 +416,12 @@ CREATE TABLE replication_metadata ( WITH (user_catalog_table = true) ; \d+ replication_metadata - Table "public.replication_metadata" - Column | Type | Collation | Nullable | Default | Storage | Stats target | Description -----------+---------+-----------+----------+--------------------------------------------------+----------+--------------+------------- - id | integer | | not null | nextval('replication_metadata_id_seq'::regclass) | plain | | - relation | name | | not null | | plain | | - options | text[] | | | | extended | | + Table "public.replication_metadata" + Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description +----------+---------+-----------+----------+--------------------------------------------------+----------+-------------+--------------+------------- + id | integer | | not null | nextval('replication_metadata_id_seq'::regclass) | plain | | | + relation | name | | not null | | plain | | | + options | text[] | | | | extended | | | Indexes: "replication_metadata_pkey" PRIMARY KEY, btree (id) Options: user_catalog_table=true @@ -430,12 +430,12 @@ INSERT INTO replication_metadata(relation, options) VALUES ('foo', ARRAY['a', 'b']); ALTER TABLE replication_metadata RESET (user_catalog_table); \d+ replication_metadata - Table "public.replication_metadata" - Column | Type | Collation | Nullable | Default | Storage | Stats target | Description -----------+---------+-----------+----------+--------------------------------------------------+----------+--------------+------------- - id | integer | | not null | nextval('replication_metadata_id_seq'::regclass) | plain | | - relation | name | | not null | | plain | | - options | text[] | | | | extended | | + Table "public.replication_metadata" + Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description +----------+---------+-----------+----------+--------------------------------------------------+----------+-------------+--------------+------------- + id | integer | | not null | nextval('replication_metadata_id_seq'::regclass) | plain | | | + relation | name | | not null | | plain | | | + options | text[] | | | | extended | | | Indexes: "replication_metadata_pkey" PRIMARY KEY, btree (id) @@ -443,12 +443,12 @@ INSERT INTO replication_metadata(relation, options) VALUES ('bar', ARRAY['a', 'b']); ALTER TABLE replication_metadata SET (user_catalog_table = true); \d+ replication_metadata - Table "public.replication_metadata" - Column | Type | Collation | Nullable | Default | Storage | Stats target | Description -----------+---------+-----------+----------+--------------------------------------------------+----------+--------------+------------- - id | integer | | not null | nextval('replication_metadata_id_seq'::regclass) | plain | | - relation | name | | not null | | plain | | - options | text[] | | | | extended | | + Table "public.replication_metadata" + Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description +----------+---------+-----------+----------+--------------------------------------------------+----------+-------------+--------------+------------- + id | integer | | not null | nextval('replication_metadata_id_seq'::regclass) | plain | | | + relation | name | | not null | | plain | | | + options | text[] | | | | extended | | | Indexes: "replication_metadata_pkey" PRIMARY KEY, btree (id) Options: user_catalog_table=true @@ -461,13 +461,13 @@ ALTER TABLE replication_metadata ALTER COLUMN rewritemeornot TYPE text; ERROR: cannot rewrite table "replication_metadata" used as a catalog table ALTER TABLE replication_metadata SET (user_catalog_table = false); \d+ replication_metadata - Table "public.replication_metadata" - Column | Type | Collation | Nullable | Default | Storage | Stats target | Description -----------------+---------+-----------+----------+--------------------------------------------------+----------+--------------+------------- - id | integer | | not null | nextval('replication_metadata_id_seq'::regclass) | plain | | - relation | name | | not null | | plain | | - options | text[] | | | | extended | | - rewritemeornot | integer | | | | plain | | + Table "public.replication_metadata" + Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description +----------------+---------+-----------+----------+--------------------------------------------------+----------+-------------+--------------+------------- + id | integer | | not null | nextval('replication_metadata_id_seq'::regclass) | plain | | | + relation | name | | not null | | plain | | | + options | text[] | | | | extended | | | + rewritemeornot | integer | | | | plain | | | Indexes: "replication_metadata_pkey" PRIMARY KEY, btree (id) Options: user_catalog_table=false diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c index 138671410a..145f59b949 100644 --- a/src/backend/access/common/indextuple.c +++ b/src/backend/access/common/indextuple.c @@ -92,7 +92,8 @@ 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], + TupleDescAttrCompression(tupleDescriptor, i)); if (DatumGetPointer(cvalue) != NULL) { diff --git a/src/backend/access/common/reloptions.c b/src/backend/access/common/reloptions.c index ec10762529..21e375ad96 100644 --- a/src/backend/access/common/reloptions.c +++ b/src/backend/access/common/reloptions.c @@ -935,11 +935,48 @@ untransformRelOptions(Datum options) val = (Node *) makeString(pstrdup(p)); } result = lappend(result, makeDefElem(pstrdup(s), val, -1)); + pfree(s); } return result; } +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; +} + +void +freeRelOptions(List *options) +{ + ListCell *cell; + + Assert(options != NIL); + foreach(cell, options) + { + DefElem *def = (DefElem *) lfirst(cell); + + pfree(def->defname); + pfree(defGetString(def)); + pfree(def->arg); + } + list_free_deep(options); +} + /* * Extract and parse reloptions from a pg_class tuple. * diff --git a/src/backend/access/common/tupdesc.c b/src/backend/access/common/tupdesc.c index 4436c86361..7e79b89beb 100644 --- a/src/backend/access/common/tupdesc.c +++ b/src/backend/access/common/tupdesc.c @@ -19,8 +19,10 @@ #include "postgres.h" +#include "access/compression.h" #include "access/hash.h" #include "access/htup_details.h" +#include "access/reloptions.h" #include "catalog/pg_collation.h" #include "catalog/pg_type.h" #include "miscadmin.h" @@ -65,6 +67,7 @@ CreateTemplateTupleDesc(int natts, bool hasoid) desc->tdtypmod = -1; desc->tdhasoid = hasoid; desc->tdrefcount = -1; /* assume not reference-counted */ + desc->tdcompression = NULL; return desc; } @@ -91,6 +94,27 @@ CreateTupleDesc(int natts, bool hasoid, Form_pg_attribute *attrs) return desc; } +static AttributeCompression * +TupleDescCopyCompression(TupleDesc tupdesc) +{ + AttributeCompression *src = tupdesc->tdcompression; + AttributeCompression *dst = (AttributeCompression *) + palloc0(sizeof(*dst) * tupdesc->natts); + int i; + + for (i = 0; i < tupdesc->natts; i++) + if (src[i].routine) + { + dst[i].routine = palloc(sizeof(*dst[i].routine)); + memcpy(dst[i].routine, src[i].routine, sizeof(*dst[i].routine)); + + dst[i].cmoptoid = src[i].cmoptoid; + dst[i].options = copyObject(src[i].options); + } + + return dst; +} + /* * CreateTupleDescCopy * This function creates a new TupleDesc by copying from an existing @@ -119,6 +143,9 @@ CreateTupleDescCopy(TupleDesc tupdesc) desc->tdtypeid = tupdesc->tdtypeid; desc->tdtypmod = tupdesc->tdtypmod; + if (tupdesc->tdcompression) + desc->tdcompression = TupleDescCopyCompression(tupdesc); + return desc; } @@ -178,6 +205,9 @@ CreateTupleDescCopyConstr(TupleDesc tupdesc) desc->constr = cpy; } + if (tupdesc->tdcompression) + desc->tdcompression = TupleDescCopyCompression(tupdesc); + desc->tdtypeid = tupdesc->tdtypeid; desc->tdtypmod = tupdesc->tdtypmod; @@ -226,6 +256,7 @@ TupleDescCopyEntry(TupleDesc dst, AttrNumber dstAttno, dstAtt->attnotnull = false; dstAtt->atthasdef = false; dstAtt->attidentity = '\0'; + dstAtt->attcompression = InvalidOid; } /* @@ -271,6 +302,23 @@ FreeTupleDesc(TupleDesc tupdesc) pfree(tupdesc->constr); } + if (tupdesc->tdcompression) + { + for (i = 0; i < tupdesc->natts; i++) + { + AttributeCompression *ac = &tupdesc->tdcompression[i]; + + if (ac->routine) + { + pfree(ac->routine); + if (ac->options) + freeRelOptions(ac->options); + } + } + + pfree(tupdesc->tdcompression); + } + pfree(tupdesc); } @@ -380,6 +428,8 @@ equalTupleDescs(TupleDesc tupdesc1, TupleDesc tupdesc2) return false; if (attr1->attcollation != attr2->attcollation) return false; + if (attr1->attcompression != attr2->attcompression) + return false; /* attacl, attoptions and attfdwoptions are not even present... */ } @@ -442,6 +492,22 @@ equalTupleDescs(TupleDesc tupdesc1, TupleDesc tupdesc2) } else if (tupdesc2->constr != NULL) return false; + + if ((tupdesc1->tdcompression == NULL) != (tupdesc2->tdcompression == NULL)) + return false; + + if (tupdesc1->tdcompression == NULL) + return true; + + for (i = 0; i < tupdesc1->natts; i++) + { + AttributeCompression *ac1 = &tupdesc1->tdcompression[i]; + AttributeCompression *ac2 = &tupdesc2->tdcompression[i]; + + if (ac1->cmoptoid != ac2->cmoptoid) + return false; + } + return true; } @@ -547,6 +613,7 @@ TupleDescInitEntry(TupleDesc desc, att->attalign = typeForm->typalign; att->attstorage = typeForm->typstorage; att->attcollation = typeForm->typcollation; + att->attcompression = InvalidOid; ReleaseSysCache(tuple); } @@ -659,6 +726,26 @@ TupleDescInitEntryCollation(TupleDesc desc, TupleDescAttr(desc, attributeNumber - 1)->attcollation = collationid; } +void +TupleDescInitAttrCompression(TupleDesc desc, + AttrNumber attnum, + Oid cmoptoid) +{ + CompressionMethodRoutine *cmr; + AttributeCompression *ac; + + cmr = GetCompressionRoutine(cmoptoid); + + /* initialize array with compression routines */ + if (!desc->tdcompression) + desc->tdcompression = (AttributeCompression *) + palloc0(desc->natts * sizeof(AttributeCompression)); + + ac = TupleDescAttrCompression(desc, attnum - 1); + ac->routine = cmr; + ac->options = GetCompressionOptionsList(cmoptoid); + ac->cmoptoid = cmoptoid; +} /* * BuildDescForRelation diff --git a/src/backend/access/heap/tuptoaster.c b/src/backend/access/heap/tuptoaster.c index 5a8f1dab83..694e16f103 100644 --- a/src/backend/access/heap/tuptoaster.c +++ b/src/backend/access/heap/tuptoaster.c @@ -30,8 +30,10 @@ #include #include +#include "access/compression.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" @@ -53,19 +55,43 @@ 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 cmoptoid; /* Oid from pg_compression_opt */ +} toast_compress_header_custom; + +#define RAWSIZEMASK (0x3FFFFFFFU) + /* * Utilities for manipulation of header information for compressed * toast entries. */ -#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 { \ + ((toast_compress_header *) (ptr))->info &= 0xC0000000; \ + ((toast_compress_header *) (ptr))->info |= ((uint32)(len) & RAWSIZEMASK); \ +} while (0) +#define TOAST_COMPRESS_SET_CMOPTOID(ptr, oid) \ + (((toast_compress_header_custom *) (ptr))->cmoptoid = (oid)) +#define TOAST_COMPRESS_SET_CUSTOM(ptr) \ +do { \ + (((toast_compress_header *) (ptr))->info |= (1 << 31)); \ + (((toast_compress_header *) (ptr))->info &= ~(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, @@ -773,7 +799,8 @@ toast_insert_or_update(Relation rel, HeapTuple newtup, HeapTuple oldtup, if (TupleDescAttr(tupleDesc, i)->attstorage == 'x') { old_value = toast_values[i]; - new_value = toast_compress_datum(old_value); + new_value = toast_compress_datum(old_value, + TupleDescAttrCompression(tupleDesc, i)); if (DatumGetPointer(new_value) != NULL) { @@ -914,7 +941,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, + TupleDescAttrCompression(tupleDesc, i)); if (DatumGetPointer(new_value) != NULL) { @@ -1229,7 +1257,9 @@ toast_flatten_tuple_to_datum(HeapTupleHeader tup, if (VARATT_IS_EXTERNAL(new_value) || VARATT_IS_COMPRESSED(new_value)) { - new_value = heap_tuple_untoast_attr(new_value); + struct varlena *untoasted_value = heap_tuple_untoast_attr(new_value); + + new_value = untoasted_value; toast_values[i] = PointerGetDatum(new_value); toast_free[i] = true; } @@ -1353,7 +1383,6 @@ toast_build_flattened_tuple(TupleDesc tupleDesc, return new_tuple; } - /* ---------- * toast_compress_datum - * @@ -1368,11 +1397,12 @@ toast_build_flattened_tuple(TupleDesc tupleDesc, * ---------- */ Datum -toast_compress_datum(Datum value) +toast_compress_datum(Datum value, AttributeCompression * ac) { struct varlena *tmp; - int32 valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value)); - int32 len; + int32 valsize, + len = 0; + bool custom = (ac != NULL); Assert(!VARATT_IS_EXTERNAL(DatumGetPointer(value))); Assert(!VARATT_IS_COMPRESSED(DatumGetPointer(value))); @@ -1381,12 +1411,27 @@ toast_compress_datum(Datum 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) + valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value)); + if (!custom && (valsize < PGLZ_strategy_default->min_input_size || + valsize > PGLZ_strategy_default->max_input_size)) return PointerGetDatum(NULL); - tmp = (struct varlena *) palloc(PGLZ_MAX_OUTPUT(valsize) + - TOAST_COMPRESS_HDRSZ); + if (custom) + { + Assert(ac->routine); + tmp = ac->routine->compress(ac, (const struct varlena *) value); + if (!tmp) + return PointerGetDatum(NULL); + } + else + { + tmp = (struct varlena *) palloc(PGLZ_MAX_OUTPUT(valsize) + + TOAST_COMPRESS_HDRSZ); + len = pglz_compress(VARDATA_ANY(DatumGetPointer(value)), + valsize, + TOAST_COMPRESS_RAWDATA(tmp), + PGLZ_strategy_default); + } /* * We recheck the actual size even if pglz_compress() reports success, @@ -1398,11 +1443,7 @@ toast_compress_datum(Datum value) * 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 && + if (!custom && len >= 0 && len + TOAST_COMPRESS_HDRSZ < valsize - 2) { TOAST_COMPRESS_SET_RAWSIZE(tmp, valsize); @@ -1410,6 +1451,14 @@ toast_compress_datum(Datum value) /* successful compression */ return PointerGetDatum(tmp); } + else if (custom && VARSIZE(tmp) < valsize - 2) + { + TOAST_COMPRESS_SET_CUSTOM(tmp); + TOAST_COMPRESS_SET_RAWSIZE(tmp, valsize); + TOAST_COMPRESS_SET_CMOPTOID(tmp, ac->cmoptoid); + /* successful compression */ + return PointerGetDatum(tmp); + } else { /* incompressible data */ @@ -2280,15 +2329,29 @@ 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 (pglz_decompress(TOAST_COMPRESS_RAWDATA(attr), - VARSIZE(attr) - TOAST_COMPRESS_HDRSZ, - VARDATA(result), - TOAST_COMPRESS_RAWSIZE(attr)) < 0) - elog(ERROR, "compressed data is corrupted"); + if (VARATT_IS_CUSTOM_COMPRESSED(attr)) + { + CompressionMethodRoutine *cmr; + toast_compress_header_custom *hdr = \ + (toast_compress_header_custom *) attr; + List *options; + + /* TODO: add some cache for options and routines */ + cmr = GetCompressionRoutine(hdr->cmoptoid); + options = GetCompressionOptionsList(hdr->cmoptoid); + result = cmr->decompress(attr, options); + } + 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; } diff --git a/src/backend/bootstrap/bootstrap.c b/src/backend/bootstrap/bootstrap.c index 0453fd4ac1..53feb17abc 100644 --- a/src/backend/bootstrap/bootstrap.c +++ b/src/backend/bootstrap/bootstrap.c @@ -718,6 +718,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 fd33426bad..c7cea974b1 100644 --- a/src/backend/catalog/Makefile +++ b/src/backend/catalog/Makefile @@ -46,7 +46,7 @@ POSTGRES_BKI_SRCS = $(addprefix $(top_srcdir)/src/include/catalog/,\ pg_collation.h pg_partitioned_table.h pg_range.h pg_transform.h \ pg_sequence.h pg_publication.h pg_publication_rel.h pg_subscription.h \ pg_subscription_rel.h toasting.h indexing.h \ - toasting.h indexing.h \ + pg_compression.h pg_compression_opt.h \ ) # location of Catalog.pm diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c index ccde66a7dd..fd733a34a0 100644 --- a/src/backend/catalog/aclchk.c +++ b/src/backend/catalog/aclchk.c @@ -3340,6 +3340,8 @@ static const char *const no_priv_msg[MAX_ACL_KIND] = gettext_noop("permission denied for publication %s"), /* ACL_KIND_SUBSCRIPTION */ gettext_noop("permission denied for subscription %s"), + /* ACL_KIND_COMPRESSION_METHOD */ + gettext_noop("permission denied for compression method %s"), }; static const char *const not_owner_msg[MAX_ACL_KIND] = diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c index 6fffc290fa..faaf7fb7f1 100644 --- a/src/backend/catalog/dependency.c +++ b/src/backend/catalog/dependency.c @@ -28,6 +28,8 @@ #include "catalog/pg_cast.h" #include "catalog/pg_collation.h" #include "catalog/pg_collation_fn.h" +#include "catalog/pg_compression.h" +#include "catalog/pg_compression_opt.h" #include "catalog/pg_constraint.h" #include "catalog/pg_constraint_fn.h" #include "catalog/pg_conversion.h" @@ -173,7 +175,9 @@ static const Oid object_classes[] = { PublicationRelationId, /* OCLASS_PUBLICATION */ PublicationRelRelationId, /* OCLASS_PUBLICATION_REL */ SubscriptionRelationId, /* OCLASS_SUBSCRIPTION */ - TransformRelationId /* OCLASS_TRANSFORM */ + TransformRelationId, /* OCLASS_TRANSFORM */ + CompressionMethodRelationId, /* OCLASS_COMPRESSION_METHOD */ + CompressionOptRelationId, /* OCLASS_COMPRESSION_OPTIONS */ }; @@ -1271,6 +1275,14 @@ doDeletion(const ObjectAddress *object, int flags) DropTransformById(object->objectId); break; + case OCLASS_COMPRESSION_METHOD: + RemoveCompressionMethodById(object->objectId); + break; + + case OCLASS_COMPRESSION_OPTIONS: + RemoveCompressionOptionsById(object->objectId); + break; + /* * These global object types are not supported here. */ @@ -2459,6 +2471,12 @@ getObjectClass(const ObjectAddress *object) case TransformRelationId: return OCLASS_TRANSFORM; + + case CompressionMethodRelationId: + return OCLASS_COMPRESSION_METHOD; + + case CompressionOptRelationId: + return OCLASS_COMPRESSION_OPTIONS; } /* shouldn't get here */ diff --git a/src/backend/catalog/genbki.pl b/src/backend/catalog/genbki.pl index 2eebb061b7..36c7fa0484 100644 --- a/src/backend/catalog/genbki.pl +++ b/src/backend/catalog/genbki.pl @@ -449,6 +449,7 @@ sub emit_pgattr_row attisdropped => 'f', attislocal => 't', attinhcount => '0', + attcompression=> '0', attacl => '_null_', attoptions => '_null_', attfdwoptions => '_null_'); diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c index 45ee9ac8b9..88fe220e2e 100644 --- a/src/backend/catalog/heap.c +++ b/src/backend/catalog/heap.c @@ -29,8 +29,10 @@ */ #include "postgres.h" +#include "access/compression.h" #include "access/htup_details.h" #include "access/multixact.h" +#include "access/reloptions.h" #include "access/sysattr.h" #include "access/transam.h" #include "access/xact.h" @@ -44,6 +46,8 @@ #include "catalog/partition.h" #include "catalog/pg_attrdef.h" #include "catalog/pg_collation.h" +#include "catalog/pg_compression.h" +#include "catalog/pg_compression_opt.h" #include "catalog/pg_constraint.h" #include "catalog/pg_constraint_fn.h" #include "catalog/pg_foreign_table.h" @@ -628,6 +632,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; @@ -707,6 +712,13 @@ AddNewAttributeTuples(Oid new_rel_oid, referenced.objectSubId = 0; recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL); } + + if (OidIsValid(attr->attcompression)) + { + ObjectAddressSet(referenced, CompressionOptRelationId, + attr->attcompression); + recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL); + } } /* @@ -1453,6 +1465,22 @@ DeleteRelationTuple(Oid relid) heap_close(pg_class_desc, RowExclusiveLock); } +static void +DropAttributeCompression(Form_pg_attribute att) +{ + CompressionMethodRoutine *cmr = GetCompressionRoutine(att->attcompression); + + if (cmr->drop) + { + bool attisdropped = att->attisdropped; + List *options = GetCompressionOptionsList(att->attcompression); + + att->attisdropped = true; + cmr->drop(att, options); + att->attisdropped = attisdropped; + } +} + /* * DeleteAttributeTuples * @@ -1483,7 +1511,14 @@ DeleteAttributeTuples(Oid relid) /* Delete all the matching tuples */ while ((atttup = systable_getnext(scan)) != NULL) + { + Form_pg_attribute att = (Form_pg_attribute) GETSTRUCT(atttup); + + if (OidIsValid(att->attcompression)) + DropAttributeCompression(att); + CatalogTupleDelete(attrel, &atttup->t_self); + } /* Clean up after the scan */ systable_endscan(scan); @@ -1576,6 +1611,8 @@ RemoveAttributeById(Oid relid, AttrNumber attnum) else { /* Dropping user attributes is lots harder */ + if (OidIsValid(attStruct->attcompression)) + DropAttributeCompression(attStruct); /* Mark the attribute as dropped */ attStruct->attisdropped = true; @@ -1597,6 +1634,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 c7b2f031f0..11dc107ab7 100644 --- a/src/backend/catalog/index.c +++ b/src/backend/catalog/index.c @@ -393,6 +393,7 @@ ConstructTupleDescriptor(Relation heapRelation, to->atttypmod = exprTypmod(indexkey); to->attislocal = true; to->attcollation = collationObjectId[i]; + to->attcompression = InvalidOid; ReleaseSysCache(tuple); @@ -471,6 +472,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 6cac2dfd1d..d2b5285a5a 100644 --- a/src/backend/catalog/objectaddress.c +++ b/src/backend/catalog/objectaddress.c @@ -29,6 +29,8 @@ #include "catalog/pg_default_acl.h" #include "catalog/pg_event_trigger.h" #include "catalog/pg_collation.h" +#include "catalog/pg_compression.h" +#include "catalog/pg_compression_opt.h" #include "catalog/pg_constraint.h" #include "catalog/pg_constraint_fn.h" #include "catalog/pg_conversion.h" @@ -490,6 +492,30 @@ static const ObjectPropertyType ObjectProperty[] = InvalidAttrNumber, /* no ACL (same as relation) */ ACL_KIND_STATISTICS, true + }, + { + CompressionMethodRelationId, + CompressionMethodOidIndexId, + COMPRESSIONMETHODOID, + COMPRESSIONMETHODNAME, + Anum_pg_compression_cmname, + InvalidAttrNumber, + InvalidAttrNumber, + InvalidAttrNumber, + -1, + true + }, + { + CompressionOptRelationId, + CompressionOptionsOidIndexId, + COMPRESSIONOPTIONSOID, + -1, + InvalidAttrNumber, + InvalidAttrNumber, + InvalidAttrNumber, + InvalidAttrNumber, + -1, + false, } }; @@ -712,6 +738,10 @@ static const struct object_type_map /* OBJECT_STATISTIC_EXT */ { "statistics object", OBJECT_STATISTIC_EXT + }, + /* OCLASS_COMPRESSION_METHOD */ + { + "compression method", OBJECT_COMPRESSION_METHOD } }; @@ -876,6 +906,7 @@ get_object_address(ObjectType objtype, Node *object, case OBJECT_ACCESS_METHOD: case OBJECT_PUBLICATION: case OBJECT_SUBSCRIPTION: + case OBJECT_COMPRESSION_METHOD: address = get_object_address_unqualified(objtype, (Value *) object, missing_ok); break; @@ -1182,6 +1213,11 @@ get_object_address_unqualified(ObjectType objtype, address.objectId = get_subscription_oid(name, missing_ok); address.objectSubId = 0; break; + case OBJECT_COMPRESSION_METHOD: + address.classId = CompressionMethodRelationId; + address.objectId = get_compression_method_oid(name, missing_ok); + address.objectSubId = 0; + break; default: elog(ERROR, "unrecognized objtype: %d", (int) objtype); /* placate compiler, which doesn't know elog won't return */ @@ -2139,6 +2175,7 @@ pg_get_object_address(PG_FUNCTION_ARGS) case OBJECT_SCHEMA: case OBJECT_SUBSCRIPTION: case OBJECT_TABLESPACE: + case OBJECT_COMPRESSION_METHOD: if (list_length(name) != 1) ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), @@ -2395,12 +2432,14 @@ check_object_ownership(Oid roleid, ObjectType objtype, ObjectAddress address, case OBJECT_TSPARSER: case OBJECT_TSTEMPLATE: case OBJECT_ACCESS_METHOD: + case OBJECT_COMPRESSION_METHOD: /* We treat these object types as being owned by superusers */ if (!superuser_arg(roleid)) ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), errmsg("must be superuser"))); break; + case OBJECT_STATISTIC_EXT: if (!pg_statistics_object_ownercheck(address.objectId, roleid)) aclcheck_error_type(ACLCHECK_NOT_OWNER, address.objectId); @@ -3398,6 +3437,27 @@ getObjectDescription(const ObjectAddress *object) break; } + case OCLASS_COMPRESSION_METHOD: + { + char *name = get_compression_method_name(object->objectId); + + if (!name) + elog(ERROR, "cache lookup failed for compression method %u", + object->objectId); + appendStringInfo(&buffer, _("compression method %s"), name); + pfree(name); + break; + } + + case OCLASS_COMPRESSION_OPTIONS: + { + char *name = get_compression_method_name_for_opt(object->objectId); + + appendStringInfo(&buffer, _("compression options for %s"), name); + pfree(name); + break; + } + case OCLASS_TRANSFORM: { HeapTuple trfTup; @@ -3919,6 +3979,14 @@ getObjectTypeDescription(const ObjectAddress *object) appendStringInfoString(&buffer, "transform"); break; + case OCLASS_COMPRESSION_METHOD: + appendStringInfoString(&buffer, "compression method"); + break; + + case OCLASS_COMPRESSION_OPTIONS: + appendStringInfoString(&buffer, "compression options"); + break; + /* * There's intentionally no default: case here; we want the * compiler to warn if a new OCLASS hasn't been handled above. @@ -4160,6 +4228,30 @@ getObjectIdentityParts(const ObjectAddress *object, break; } + case OCLASS_COMPRESSION_METHOD: + { + char *cmname = get_compression_method_name(object->objectId); + + if (!cmname) + elog(ERROR, "cache lookup failed for compression method %u", + object->objectId); + appendStringInfoString(&buffer, quote_identifier(cmname)); + if (objname) + *objname = list_make1(cmname); + else + pfree(cmname); + break; + } + + case OCLASS_COMPRESSION_OPTIONS: + { + appendStringInfo(&buffer, "%u", + object->objectId); + if (objname) + *objname = list_make1(psprintf("%u", object->objectId)); + break; + } + case OCLASS_CONSTRAINT: { HeapTuple conTup; diff --git a/src/backend/catalog/pg_type.c b/src/backend/catalog/pg_type.c index 59ffd2104d..6e806b3334 100644 --- a/src/backend/catalog/pg_type.c +++ b/src/backend/catalog/pg_type.c @@ -22,6 +22,7 @@ #include "catalog/indexing.h" #include "catalog/objectaccess.h" #include "catalog/pg_collation.h" +#include "catalog/pg_compression.h" #include "catalog/pg_namespace.h" #include "catalog/pg_proc.h" #include "catalog/pg_type.h" @@ -118,6 +119,7 @@ TypeShellMake(const char *typeName, Oid typeNamespace, Oid ownerId) values[Anum_pg_type_typtypmod - 1] = Int32GetDatum(-1); values[Anum_pg_type_typndims - 1] = Int32GetDatum(0); values[Anum_pg_type_typcollation - 1] = ObjectIdGetDatum(InvalidOid); + values[Anum_pg_type_typdefaultcm - 1] = ObjectIdGetDatum(InvalidOid); nulls[Anum_pg_type_typdefaultbin - 1] = true; nulls[Anum_pg_type_typdefault - 1] = true; nulls[Anum_pg_type_typacl - 1] = true; @@ -362,6 +364,7 @@ TypeCreate(Oid newTypeOid, values[Anum_pg_type_typtypmod - 1] = Int32GetDatum(typeMod); values[Anum_pg_type_typndims - 1] = Int32GetDatum(typNDims); values[Anum_pg_type_typcollation - 1] = ObjectIdGetDatum(typeCollation); + values[Anum_pg_type_typdefaultcm - 1] = ObjectIdGetDatum(InvalidOid); /* * initialize the default binary value for this type. Check for nulls of 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 4f8147907c..4f18e4083f 100644 --- a/src/backend/commands/alter.c +++ b/src/backend/commands/alter.c @@ -385,6 +385,7 @@ ExecRenameStmt(RenameStmt *stmt) case OBJECT_TSTEMPLATE: case OBJECT_PUBLICATION: case OBJECT_SUBSCRIPTION: + case OBJECT_COMPRESSION_METHOD: { ObjectAddress address; Relation catalog; @@ -500,6 +501,7 @@ ExecAlterObjectSchemaStmt(AlterObjectSchemaStmt *stmt, case OBJECT_TSDICTIONARY: case OBJECT_TSPARSER: case OBJECT_TSTEMPLATE: + case OBJECT_COMPRESSION_METHOD: { Relation catalog; Relation relation; @@ -627,6 +629,8 @@ AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid, case OCLASS_PUBLICATION_REL: case OCLASS_SUBSCRIPTION: case OCLASS_TRANSFORM: + case OCLASS_COMPRESSION_METHOD: + case OCLASS_COMPRESSION_OPTIONS: /* ignore object types that don't have schema-qualified names */ break; diff --git a/src/backend/commands/compressioncmds.c b/src/backend/commands/compressioncmds.c new file mode 100644 index 0000000000..32cbcc3efe --- /dev/null +++ b/src/backend/commands/compressioncmds.c @@ -0,0 +1,485 @@ +/*------------------------------------------------------------------------- + * + * compressioncmds.c + * Routines for SQL commands that manipulate compression methods. + * + * Portions Copyright (c) 1996-2016, 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/heapam.h" +#include "access/htup_details.h" +#include "access/compression.h" +#include "access/reloptions.h" +#include "catalog/dependency.h" +#include "catalog/indexing.h" +#include "catalog/pg_compression.h" +#include "catalog/pg_compression_opt.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/lsyscache.h" +#include "utils/rel.h" +#include "utils/syscache.h" + +static CompressionMethodRoutine *get_compression_method_routine(Oid cmhandler, Oid typeid); + +/* + * Convert a handler function name to an Oid. If the return type of the + * function doesn't match the given AM type, an error is raised. + * + * This function either return valid function Oid or throw an error. + */ +static Oid +LookupCompressionHandlerFunc(List *handlerName) +{ + static const Oid funcargtypes[1] = {INTERNALOID}; + Oid handlerOid; + + if (handlerName == NIL) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_FUNCTION), + errmsg("handler function is not specified"))); + + /* handlers have one argument of type internal */ + handlerOid = LookupFuncName(handlerName, 1, funcargtypes, false); + + /* check that handler has the correct return type */ + if (get_func_rettype(handlerOid) != COMPRESSION_HANDLEROID) + ereport(ERROR, + (errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("function %s must return type %s", + NameListToString(handlerName), + "compression_handler"))); + + return handlerOid; +} + +static ObjectAddress +CreateCompressionMethod(char *cmName, List *handlerName) +{ + Relation rel; + ObjectAddress myself; + ObjectAddress referenced; + Oid cmoid; + Oid cmhandler; + bool nulls[Natts_pg_compression]; + Datum values[Natts_pg_compression]; + HeapTuple tup; + + rel = heap_open(CompressionMethodRelationId, RowExclusiveLock); + + /* Check if name is used */ + cmoid = GetSysCacheOid1(COMPRESSIONMETHODNAME, CStringGetDatum(cmName)); + if (OidIsValid(cmoid)) + ereport(ERROR, + (errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("compression method \"%s\" already exists", + cmName))); + + /* + * Get the handler function oid and compression method routine + */ + cmhandler = LookupCompressionHandlerFunc(handlerName); + + /* + * Insert tuple into pg_compression. + */ + memset(values, 0, sizeof(values)); + memset(nulls, false, sizeof(nulls)); + + values[Anum_pg_compression_cmname - 1] = + DirectFunctionCall1(namein, CStringGetDatum(cmName)); + values[Anum_pg_compression_cmhandler - 1] = ObjectIdGetDatum(cmhandler); + + tup = heap_form_tuple(RelationGetDescr(rel), values, nulls); + + cmoid = CatalogTupleInsert(rel, tup); + heap_freetuple(tup); + + ObjectAddressSet(myself, CompressionMethodRelationId, cmoid); + + /* Record dependency on handler function */ + ObjectAddressSet(referenced, ProcedureRelationId, cmhandler); + + recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL); + recordDependencyOnCurrentExtension(&myself, false); + + heap_close(rel, RowExclusiveLock); + + return myself; +} + +ObjectAddress +DefineCompressionMethod(List *names, List *parameters) +{ + char *cmName; + ListCell *pl; + DefElem *handlerEl = NULL; + + if (list_length(names) != 1) + elog(ERROR, "compression method name cannot be qualified"); + + cmName = strVal(linitial(names)); + + /* Must be super user */ + if (!superuser()) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("permission denied to create compression method \"%s\"", + cmName), + errhint("Must be superuser to create an compression method."))); + + foreach(pl, parameters) + { + DefElem *defel = (DefElem *) lfirst(pl); + DefElem **defelp; + + if (pg_strcasecmp(defel->defname, "handler") == 0) + defelp = &handlerEl; + else + { + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("compression method attribute \"%s\" not recognized", + defel->defname))); + break; + } + + *defelp = defel; + } + + if (!handlerEl) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("compression method handler is not specified"))); + + return CreateCompressionMethod(cmName, (List *) handlerEl->arg); +} + +void +RemoveCompressionMethodById(Oid cmOid) +{ + Relation relation; + HeapTuple tup; + + if (!superuser()) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("must be superuser to drop compression methods"))); + + relation = heap_open(CompressionMethodRelationId, RowExclusiveLock); + + tup = SearchSysCache1(COMPRESSIONMETHODOID, ObjectIdGetDatum(cmOid)); + if (!HeapTupleIsValid(tup)) + elog(ERROR, "cache lookup failed for compression method %u", cmOid); + + CatalogTupleDelete(relation, &tup->t_self); + + ReleaseSysCache(tup); + + heap_close(relation, RowExclusiveLock); +} + +/* + * Create new record in pg_compression_opt + */ +Oid +CreateCompressionOptions(Form_pg_attribute attr, Oid cmid, List *options) +{ + Relation rel; + HeapTuple tuple; + Oid result, + cmhandler; + Datum values[Natts_pg_compression_opt]; + bool nulls[Natts_pg_compression_opt]; + ObjectAddress myself, + ref1, + ref2, + ref3; + CompressionMethodRoutine *routine; + + /* Initialize buffers for new tuple values */ + memset(values, 0, sizeof(values)); + memset(nulls, false, sizeof(nulls)); + + /* Get handler function OID for the compression method */ + tuple = SearchSysCache1(COMPRESSIONMETHODOID, ObjectIdGetDatum(cmid)); + if (!HeapTupleIsValid(tuple)) + elog(ERROR, "cache lookup failed for compression method %u", cmid); + cmhandler = ((Form_pg_compression) GETSTRUCT(tuple))->cmhandler; + ReleaseSysCache(tuple); + + routine = get_compression_method_routine(cmhandler, attr->atttypid); + + if (routine->configure && options != NIL) + routine->configure(attr, options); + + values[Anum_pg_compression_opt_cmid - 1] = ObjectIdGetDatum(cmid); + values[Anum_pg_compression_opt_cmhandler - 1] = ObjectIdGetDatum(cmhandler); + + if (options) + values[Anum_pg_compression_opt_cmoptions - 1] = optionListToArray(options); + else + nulls[Anum_pg_compression_opt_cmoptions - 1] = true; + + rel = heap_open(CompressionOptRelationId, RowExclusiveLock); + tuple = heap_form_tuple(RelationGetDescr(rel), values, nulls); + result = CatalogTupleInsert(rel, tuple); + heap_freetuple(tuple); + + ObjectAddressSet(myself, CompressionOptRelationId, result); + ObjectAddressSet(ref1, ProcedureRelationId, cmhandler); + ObjectAddressSubSet(ref2, RelationRelationId, attr->attrelid, attr->attnum); + ObjectAddressSet(ref3, CompressionMethodRelationId, cmid); + + recordDependencyOn(&myself, &ref1, DEPENDENCY_NORMAL); + recordDependencyOn(&ref2, &myself, DEPENDENCY_NORMAL); + recordDependencyOn(&myself, &ref3, DEPENDENCY_NORMAL); + recordDependencyOnCurrentExtension(&myself, false); + heap_close(rel, RowExclusiveLock); + + return result; +} + +void +RemoveCompressionOptionsById(Oid cmoptoid) +{ + Relation relation; + HeapTuple tup; + + if (!superuser()) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("must be superuser to drop compression options"))); + + relation = heap_open(CompressionOptRelationId, RowExclusiveLock); + + tup = SearchSysCache1(COMPRESSIONOPTIONSOID, ObjectIdGetDatum(cmoptoid)); + if (!HeapTupleIsValid(tup)) + elog(ERROR, "cache lookup failed for compression options %u", cmoptoid); + + CatalogTupleDelete(relation, &tup->t_self); + ReleaseSysCache(tup); + heap_close(relation, RowExclusiveLock); +} + +ColumnCompression * +GetColumnCompressionForAttribute(Form_pg_attribute att) +{ + HeapTuple tuple; + Form_pg_compression_opt cmopt; + ColumnCompression *compression = makeNode(ColumnCompression); + + /* Get handler function OID for the compression method */ + tuple = SearchSysCache1(COMPRESSIONOPTIONSOID, + ObjectIdGetDatum(att->attcompression)); + + if (!HeapTupleIsValid(tuple)) + elog(ERROR, "cache lookup failed for compression options %u", + att->attcompression); + + cmopt = (Form_pg_compression_opt) GETSTRUCT(tuple); + compression->methodName = NULL; + compression->methodOid = cmopt->cmid; + compression->options = GetCompressionOptionsList(att->attcompression); + ReleaseSysCache(tuple); + + return compression; +} + +void +CheckCompressionMismatch(ColumnCompression * c1, ColumnCompression * c2, + const char *attributeName) +{ + char *cmname1 = c1->methodName ? c1->methodName : + get_compression_method_name(c1->methodOid); + char *cmname2 = c2->methodName ? c2->methodName : + get_compression_method_name(c2->methodOid); + + if (strcmp(cmname1, cmname2)) + ereport(ERROR, + (errcode(ERRCODE_DATATYPE_MISMATCH), + errmsg("column \"%s\" has a compression method conflict", + attributeName), + errdetail("%s versus %s", cmname1, cmname2))); + + 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)))); +} + +/* + * get_compression_method_oid + * + * If missing_ok is false, throw an error if compression method not found. + * If missing_ok is true, just return InvalidOid. + */ +Oid +get_compression_method_oid(const char *cmname, bool missing_ok) +{ + HeapTuple tup; + Oid oid = InvalidOid; + + tup = SearchSysCache1(COMPRESSIONMETHODNAME, CStringGetDatum(cmname)); + if (HeapTupleIsValid(tup)) + { + oid = HeapTupleGetOid(tup); + ReleaseSysCache(tup); + } + + if (!OidIsValid(oid) && !missing_ok) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_OBJECT), + errmsg("compression method \"%s\" does not exist", cmname))); + + return oid; +} + +/* + * get_compression_method_name + * + * given an compression method OID, look up its name. + */ +char * +get_compression_method_name(Oid cmOid) +{ + HeapTuple tup; + char *result = NULL; + + tup = SearchSysCache1(COMPRESSIONMETHODOID, ObjectIdGetDatum(cmOid)); + if (HeapTupleIsValid(tup)) + { + Form_pg_compression cmform = (Form_pg_compression) GETSTRUCT(tup); + + result = pstrdup(NameStr(cmform->cmname)); + ReleaseSysCache(tup); + } + return result; +} + +/* + * get_compression_method_name_for_opt + * + * given an compression options OID, look up a name for used compression method. + */ +char * +get_compression_method_name_for_opt(Oid cmoptoid) +{ + HeapTuple tup; + Oid cmid; + char *result = NULL; + Form_pg_compression cmform; + + tup = SearchSysCache1(COMPRESSIONOPTIONSOID, ObjectIdGetDatum(cmoptoid)); + + if (!HeapTupleIsValid(tup)) + elog(ERROR, "cache lookup failed for compression options %u", cmoptoid); + + cmid = ((Form_pg_compression_opt) GETSTRUCT(tup))->cmid; + ReleaseSysCache(tup); + + /* now we can get the name */ + tup = SearchSysCache1(COMPRESSIONMETHODOID, ObjectIdGetDatum(cmid)); + if (!HeapTupleIsValid(tup)) + elog(ERROR, "cache lookup failed for compression method %u", cmid); + + cmform = (Form_pg_compression) GETSTRUCT(tup); + result = pstrdup(NameStr(cmform->cmname)); + ReleaseSysCache(tup); + return result; +} + +/* get_compression_options */ +List * +GetCompressionOptionsList(Oid cmoptoid) +{ + HeapTuple tuple; + List *result = NULL; + bool isnull; + Datum cmoptions; + + /* Get handler function OID for the compression method */ + tuple = SearchSysCache1(COMPRESSIONOPTIONSOID, ObjectIdGetDatum(cmoptoid)); + + if (!HeapTupleIsValid(tuple)) + elog(ERROR, "cache lookup failed for compression options %u", cmoptoid); + + cmoptions = SysCacheGetAttr(COMPRESSIONOPTIONSOID, tuple, + Anum_pg_compression_opt_cmoptions, &isnull); + + if (!isnull) + result = untransformRelOptions(cmoptions); + + ReleaseSysCache(tuple); + return result; +} + +CompressionMethodRoutine * +GetCompressionRoutine(Oid cmoptoid) +{ + HeapTuple tuple; + Form_pg_compression_opt cmopt; + regproc cmhandler; + + /* Get handler function OID for the compression method */ + tuple = SearchSysCache1(COMPRESSIONOPTIONSOID, ObjectIdGetDatum(cmoptoid)); + + if (!HeapTupleIsValid(tuple)) + elog(ERROR, "cache lookup failed for compression options %u", cmoptoid); + + cmopt = (Form_pg_compression_opt) GETSTRUCT(tuple); + cmhandler = cmopt->cmhandler; + + /* Complain if handler OID is invalid */ + if (!RegProcedureIsValid(cmhandler)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not find compression method handler for compression options %u", + cmoptoid))); + + ReleaseSysCache(tuple); + + /* And finally, call the handler function to get the API struct. */ + return get_compression_method_routine(cmhandler, InvalidOid); +} + +/* + * Call the specified compression method handler + * routine to get its CompressionMethodRoutine struct, + * which will be palloc'd in the caller's context. + */ +static CompressionMethodRoutine * +get_compression_method_routine(Oid cmhandler, Oid typeid) +{ + Datum datum; + CompressionMethodRoutine *routine; + CompressionMethodOpArgs opargs; + + opargs.typeid = typeid; + opargs.cmhanderid = cmhandler; + + datum = OidFunctionCall1(cmhandler, PointerGetDatum(&opargs)); + routine = (CompressionMethodRoutine *) DatumGetPointer(datum); + + if (routine == NULL || !IsA(routine, CompressionMethodRoutine)) + elog(ERROR, "compression method handler function %u " + "did not return an CompressionMethodRoutine struct", + cmhandler); + + return routine; +} diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index cfa3f059c2..8cef3be086 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -2758,8 +2758,8 @@ CopyFrom(CopyState cstate) List *recheckIndexes = NIL; /* OK, store the tuple and create index entries for it */ - heap_insert(resultRelInfo->ri_RelationDesc, tuple, mycid, - hi_options, bistate); + heap_insert(resultRelInfo->ri_RelationDesc, tuple, + mycid, hi_options, bistate); if (resultRelInfo->ri_NumIndices > 0) recheckIndexes = ExecInsertIndexTuples(slot, diff --git a/src/backend/commands/createas.c b/src/backend/commands/createas.c index e60210cb24..010bdb644b 100644 --- a/src/backend/commands/createas.c +++ b/src/backend/commands/createas.c @@ -112,7 +112,8 @@ create_ctas_internal(List *attrList, IntoClause *into) * Create the relation. (This will error out if there's an existing view, * so we don't need more code to complain if "replace" is false.) */ - intoRelationAddr = DefineRelation(create, relkind, InvalidOid, NULL, NULL); + intoRelationAddr = DefineRelation(create, relkind, InvalidOid, NULL, NULL, + NULL); /* * If necessary, create a TOAST table for the target table. Note that diff --git a/src/backend/commands/dropcmds.c b/src/backend/commands/dropcmds.c index 2b30677d6f..8611db22ec 100644 --- a/src/backend/commands/dropcmds.c +++ b/src/backend/commands/dropcmds.c @@ -275,6 +275,10 @@ does_not_exist_skipping(ObjectType objtype, Node *object) name = NameListToString(castNode(List, object)); } break; + case OBJECT_COMPRESSION_METHOD: + msg = gettext_noop("compression method \"%s\" does not exist, skipping"); + name = strVal((Value *) object); + break; case OBJECT_CONVERSION: if (!schema_does_not_exist_skipping(castNode(List, object), &msg, &name)) { diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c index 938133bbe4..f520c7fe7b 100644 --- a/src/backend/commands/event_trigger.c +++ b/src/backend/commands/event_trigger.c @@ -91,6 +91,7 @@ static event_trigger_support_data event_trigger_support[] = { {"CAST", true}, {"CONSTRAINT", true}, {"COLLATION", true}, + {"COMPRESSION METHOD", true}, {"CONVERSION", true}, {"DATABASE", false}, {"DOMAIN", true}, @@ -1085,6 +1086,7 @@ EventTriggerSupportsObjectType(ObjectType obtype) case OBJECT_CAST: case OBJECT_COLUMN: case OBJECT_COLLATION: + case OBJECT_COMPRESSION_METHOD: case OBJECT_CONVERSION: case OBJECT_DEFACL: case OBJECT_DEFAULT: @@ -1144,6 +1146,7 @@ EventTriggerSupportsObjectClass(ObjectClass objclass) case OCLASS_DATABASE: case OCLASS_TBLSPACE: case OCLASS_ROLE: + case OCLASS_COMPRESSION_OPTIONS: /* no support for global objects */ return false; case OCLASS_EVENT_TRIGGER: @@ -1183,6 +1186,7 @@ EventTriggerSupportsObjectClass(ObjectClass objclass) case OCLASS_PUBLICATION_REL: case OCLASS_SUBSCRIPTION: case OCLASS_TRANSFORM: + case OCLASS_COMPRESSION_METHOD: return true; /* diff --git a/src/backend/commands/foreigncmds.c b/src/backend/commands/foreigncmds.c index 9ad991507f..b840309d03 100644 --- a/src/backend/commands/foreigncmds.c +++ b/src/backend/commands/foreigncmds.c @@ -61,7 +61,7 @@ static void import_error_callback(void *arg); * processing, hence any validation should be done before this * conversion. */ -static Datum +Datum optionListToArray(List *options) { ArrayBuildState *astate = NULL; diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c index 62937124ef..857eb14b64 100644 --- a/src/backend/commands/sequence.c +++ b/src/backend/commands/sequence.c @@ -212,7 +212,8 @@ DefineSequence(ParseState *pstate, CreateSeqStmt *seq) stmt->tablespacename = NULL; stmt->if_not_exists = seq->if_not_exists; - address = DefineRelation(stmt, RELKIND_SEQUENCE, seq->ownerId, NULL, NULL); + address = DefineRelation(stmt, RELKIND_SEQUENCE, seq->ownerId, NULL, NULL, + NULL); seqoid = address.objectId; Assert(seqoid != InvalidOid); diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index c8fc9cb7fe..a715093a66 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -14,6 +14,7 @@ */ #include "postgres.h" +#include "access/compression.h" #include "access/genam.h" #include "access/heapam.h" #include "access/multixact.h" @@ -33,6 +34,8 @@ #include "catalog/partition.h" #include "catalog/pg_am.h" #include "catalog/pg_collation.h" +#include "catalog/pg_compression.h" +#include "catalog/pg_compression_opt.h" #include "catalog/pg_constraint.h" #include "catalog/pg_constraint_fn.h" #include "catalog/pg_depend.h" @@ -41,6 +44,7 @@ #include "catalog/pg_inherits_fn.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" @@ -90,6 +94,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" @@ -456,6 +461,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 void ATExecAlterColumnCompression(AlteredTableInfo *tab, Relation rel, + const char *column, ColumnCompression * compression, LOCKMODE lockmode); static void copy_relation_data(SMgrRelation rel, SMgrRelation dst, ForkNumber forkNum, char relpersistence); @@ -501,7 +508,8 @@ static ObjectAddress ATExecDetachPartition(Relation rel, RangeVar *name); */ ObjectAddress DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, - ObjectAddress *typaddress, const char *queryString) + ObjectAddress *typaddress, const char *queryString, + Node **pAlterStmt) { char relname[NAMEDATALEN]; Oid namespaceId; @@ -521,6 +529,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, static char *validnsps[] = HEAP_RELOPT_NAMESPACES; Oid ofTypeId; ObjectAddress address; + AlterTableStmt *alterStmt = NULL; /* * Truncate relname to appropriate length (probably a waste of time, as @@ -722,6 +731,8 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, if (colDef->identity) attr->attidentity = colDef->identity; + + transformColumnCompression(colDef, stmt->relation, &alterStmt); } /* @@ -876,6 +887,11 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, */ relation_close(rel, NoLock); + if (pAlterStmt) + *pAlterStmt = (Node *) alterStmt; + else + Assert(!alterStmt); + return address; } @@ -1565,6 +1581,7 @@ storage_name(char c) } } + /*---------- * MergeAttributes * Returns new schema given initial schema and superclasses. @@ -1899,6 +1916,19 @@ MergeAttributes(List *schema, List *supers, char relpersistence, storage_name(def->storage), storage_name(attribute->attstorage)))); + if (OidIsValid(attribute->attcompression)) + { + ColumnCompression *compression = + GetColumnCompressionForAttribute(attribute); + + 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; @@ -1926,6 +1956,9 @@ MergeAttributes(List *schema, List *supers, char relpersistence, def->collOid = attribute->attcollation; def->constraints = NIL; def->location = -1; + if (OidIsValid(attribute->attcompression)) + def->compression = + GetColumnCompressionForAttribute(attribute); inhSchema = lappend(inhSchema, def); newattno[parent_attno - 1] = ++child_attno; } @@ -2135,6 +2168,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 */ @@ -3234,6 +3274,7 @@ AlterTableGetLockLevel(List *cmds) */ case AT_GenericOptions: case AT_AlterColumnGenericOptions: + case AT_AlterColumnCompression: cmd_lockmode = AccessExclusiveLock; break; @@ -3724,6 +3765,12 @@ ATPrepCmd(List **wqueue, Relation rel, AlterTableCmd *cmd, /* No command-specific prep needed */ pass = AT_PASS_MISC; break; + case AT_AlterColumnCompression: + ATSimplePermissions(rel, ATT_TABLE); + /* FIXME This command never recurses */ + /* No command-specific prep needed */ + pass = AT_PASS_MISC; + break; default: /* oops */ elog(ERROR, "unrecognized alter table type: %d", (int) cmd->subtype); @@ -4063,6 +4110,11 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel, case AT_DetachPartition: ATExecDetachPartition(rel, ((PartitionCmd *) cmd->def)->name); break; + case AT_AlterColumnCompression: + ATExecAlterColumnCompression(tab, rel, cmd->name, + (ColumnCompression *) cmd->def, + lockmode); + break; default: /* oops */ elog(ERROR, "unrecognized alter table type: %d", (int) cmd->subtype); @@ -5274,6 +5326,18 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel, attribute.attislocal = colDef->is_local; attribute.attinhcount = colDef->inhcount; attribute.attcollation = collOid; + attribute.attcompression = InvalidOid; + + if (!colDef->compression) + { + /* colDef->compression is handled in subsequent ALTER TABLE statement */ + Oid cmid = get_base_typdefaultcm(typeTuple); + + if (OidIsValid(cmid)) + attribute.attcompression = CreateCompressionOptions(&attribute, + cmid, NULL); + } + /* attribute.attacl is handled by InsertPgAttributeTuple */ ReleaseSysCache(typeTuple); @@ -6139,8 +6203,8 @@ ATPrepSetStatistics(Relation rel, const char *colName, int16 colNum, Node *newVa RelationGetRelationName(rel)))); /* - * We allow referencing columns by numbers only for indexes, since - * table column numbers could contain gaps if columns are later dropped. + * We allow referencing columns by numbers only for indexes, since table + * column numbers could contain gaps if columns are later dropped. */ if (rel->rd_rel->relkind != RELKIND_INDEX && !colName) ereport(ERROR, @@ -6374,6 +6438,11 @@ ATExecSetStorage(Relation rel, const char *colName, Node *newValue, LOCKMODE loc errmsg("cannot alter system column \"%s\"", colName))); + if (attrtuple->attcompression && newstorage != 'x' && newstorage != 'm') + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("compressed columns can only have storage MAIN or EXTENDED"))); + /* * safety check: do not allow toasted storage modes unless column datatype * is TOAST-aware. @@ -9019,6 +9088,7 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel, SysScanDesc scan; HeapTuple depTup; ObjectAddress address; + Oid cmid; attrelation = heap_open(AttributeRelationId, RowExclusiveLock); @@ -9294,6 +9364,8 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel, case OCLASS_PUBLICATION_REL: case OCLASS_SUBSCRIPTION: case OCLASS_TRANSFORM: + case OCLASS_COMPRESSION_METHOD: + case OCLASS_COMPRESSION_OPTIONS: /* * We don't expect any of these sorts of objects to depend on @@ -9343,7 +9415,9 @@ 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 == CompressionMethodRelationId && + foundDep->refobjid == attTup->attcompression)) elog(ERROR, "found unexpected dependency for column"); CatalogTupleDelete(depRel, &depTup->t_self); @@ -9366,6 +9440,10 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel, attTup->attalign = tform->typalign; attTup->attstorage = tform->typstorage; + cmid = get_base_typdefaultcm(typeTuple); + if (OidIsValid(cmid)) + attTup->attcompression = CreateCompressionOptions(attTup, cmid, NULL); + ReleaseSysCache(typeTuple); CatalogTupleUpdate(attrelation, &heapTup->t_self, heapTup); @@ -12362,6 +12440,86 @@ ATExecGenericOptions(Relation rel, List *options) heap_freetuple(tuple); } +static void +ATExecAlterColumnCompression(AlteredTableInfo *tab, + Relation rel, + const char *column, + ColumnCompression * compression, + LOCKMODE lockmode) +{ + Relation attrel; + HeapTuple atttuple; + Form_pg_attribute atttableform; + AttrNumber attnum; + HeapTuple newtuple; + Datum values[Natts_pg_attribute]; + bool nulls[Natts_pg_attribute]; + bool replace[Natts_pg_attribute]; + + 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))); + + if (atttableform->attstorage != 'x' && atttableform->attstorage != 'm') + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("storage for \"%s\" should be MAIN or EXTENDED", column))); + + /* Initialize buffers for new tuple values */ + memset(values, 0, sizeof(values)); + memset(nulls, false, sizeof(nulls)); + memset(replace, false, sizeof(replace)); + + if (compression->methodName || OidIsValid(compression->methodOid)) + { + /* SET COMPRESSED */ + Oid cmid, + cmoptoid; + + cmid = compression->methodName + ? get_compression_method_oid(compression->methodName, false) + : compression->methodOid; + + cmoptoid = CreateCompressionOptions(atttableform, cmid, compression->options); + + values[Anum_pg_attribute_attcompression - 1] = ObjectIdGetDatum(cmoptoid); + replace[Anum_pg_attribute_attcompression - 1] = true; + + } + else + { + /* SET NOT COMPRESSED */ + values[Anum_pg_attribute_attcompression - 1] = ObjectIdGetDatum(InvalidOid); + replace[Anum_pg_attribute_attcompression - 1] = true; + } + + newtuple = heap_modify_tuple(atttuple, RelationGetDescr(attrel), + values, nulls, replace); + CatalogTupleUpdate(attrel, &newtuple->t_self, newtuple); + heap_freetuple(newtuple); + + InvokeObjectPostAlterHook(RelationRelationId, + RelationGetRelid(rel), + atttableform->attnum); + + ReleaseSysCache(atttuple); + heap_close(attrel, RowExclusiveLock); +} + + /* * Preparation phase for SET LOGGED/UNLOGGED * diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c index 7ed16aeff4..2d39f57ab0 100644 --- a/src/backend/commands/typecmds.c +++ b/src/backend/commands/typecmds.c @@ -40,6 +40,7 @@ #include "catalog/pg_am.h" #include "catalog/pg_authid.h" #include "catalog/pg_collation.h" +#include "catalog/pg_compression.h" #include "catalog/pg_constraint.h" #include "catalog/pg_constraint_fn.h" #include "catalog/pg_depend.h" @@ -2110,7 +2111,7 @@ DefineCompositeType(RangeVar *typevar, List *coldeflist) * Finally create the relation. This also creates the type. */ DefineRelation(createStmt, RELKIND_COMPOSITE_TYPE, InvalidOid, &address, - NULL); + NULL, NULL); return address; } @@ -3638,3 +3639,95 @@ AlterTypeNamespaceInternal(Oid typeOid, Oid nspOid, return oldNspOid; } + + +/* + * Execute ALTER TYPE SET COMPRESSED [WITH (