From 1880fe2b337cfa0140136f7c5b600d435afde71c Mon Sep 17 00:00:00 2001 From: dilipkumar Date: Thu, 15 Oct 2020 14:09:13 +0530 Subject: [PATCH v7 4/5] Create custom compression methods Provide syntax to create custom compression methods. --- .../sgml/ref/create_compression_method.sgml | 87 +++++++++++++++ src/backend/access/common/detoast.c | 54 +++++++-- src/backend/access/common/toast_internals.c | 15 ++- .../access/compression/compressionapi.c | 2 +- src/backend/access/compression/pglz.c | 20 ++-- src/backend/access/compression/zlib.c | 14 +-- src/backend/catalog/aclchk.c | 2 + src/backend/catalog/dependency.c | 2 +- src/backend/catalog/objectaddress.c | 22 ++++ src/backend/commands/compressioncmds.c | 103 ++++++++++++++++++ src/backend/commands/event_trigger.c | 3 + src/backend/commands/seclabel.c | 1 + src/backend/executor/nodeModifyTable.c | 3 +- src/backend/nodes/copyfuncs.c | 14 +++ src/backend/nodes/equalfuncs.c | 12 ++ src/backend/parser/gram.y | 20 +++- src/backend/tcop/utility.c | 16 +++ src/include/access/compressionapi.h | 16 ++- src/include/access/detoast.h | 8 ++ src/include/access/toast_internals.h | 19 +++- src/include/commands/defrem.h | 2 + src/include/nodes/nodes.h | 3 +- src/include/nodes/parsenodes.h | 12 ++ src/include/postgres.h | 8 ++ src/include/tcop/cmdtaglist.h | 2 + src/test/regress/expected/create_cm.out | 18 +++ src/test/regress/sql/create_cm.sql | 9 ++ 27 files changed, 445 insertions(+), 42 deletions(-) create mode 100644 doc/src/sgml/ref/create_compression_method.sgml diff --git a/doc/src/sgml/ref/create_compression_method.sgml b/doc/src/sgml/ref/create_compression_method.sgml new file mode 100644 index 0000000000..e9d076194b --- /dev/null +++ b/doc/src/sgml/ref/create_compression_method.sgml @@ -0,0 +1,87 @@ + + + + + CREATE COMPRESSION METHOD + + + + CREATE COMPRESSION METHOD + 7 + SQL - Language Statements + + + + CREATE COMPRESSION METHOD + define a new compresion method + + + + +CREATE COMPRESSION METHOD name + HANDLER handler_function + + + + + Description + + + CREATE COMPRESSION METHOD creates a new compression method. + PostgreSQL supports two internal + built-in compression methods (pglz + and zlib), and also allows to add more custom compression + methods through this interface. + + + + + The compression method name must be unique within the database. + + + + Only superusers can define new compression methods. + + + + + Parameters + + + + name + + + The name of the compression method to be created. + + + + + + handler_function + + + handler_function is the + name (possibly schema-qualified) of a previously registered function + that represents the access method. The handler function must be declared + to accept a single argument of type internal and to return + the pseudo-type compression_handler. The argument is a + dummy value that simply serves to prevent handler functions from being + called directly from SQL commands. The result of the function must be a + palloc'd struct of type CompressionRoutine, + which contains everything that the core code needs to know to make use of + the compression access method. + The CompressionRoutine struct, also called the + access method's API struct, contains pointers to + support functions for the compression method. These support functions are + plain C functions and are not visible or callable at the SQL level. + + + + + + + diff --git a/src/backend/access/common/detoast.c b/src/backend/access/common/detoast.c index 1460fbc431..6988a496cd 100644 --- a/src/backend/access/common/detoast.c +++ b/src/backend/access/common/detoast.c @@ -438,6 +438,35 @@ toast_fetch_datum_slice(struct varlena *attr, int32 sliceoffset, return result; } +/* ---------- + * toast_get_compression_oid - + * + * Return the Oid of the compresion method stored in the compressed attribute. + */ +Oid +toast_get_compression_oid(struct varlena *attr) +{ + CompressionId cmid; + + /* + * Get the compression id from the toast header and check whether the + * value is custom compressed or not. If it is custom compressed then + * fetch the oid from the custome compression header. Otherwise it is + * compressed with the built-in compression method so get the oid of the + * built-in compression methods. + */ + cmid = TOAST_COMPRESS_METHOD(attr); + if (IsCustomeCompressionID(cmid)) + { + toast_compress_header_custom *hdr; + + hdr = (toast_compress_header_custom *) attr; + return hdr->cmoid; + } + else + return GetCompressionOidFromCompressionId(cmid); +} + /* ---------- * toast_decompress_datum - * @@ -447,16 +476,20 @@ static struct varlena * toast_decompress_datum(struct varlena *attr) { CompressionRoutine *cmroutine; - Oid cmoid; + Oid cmoid; Assert(VARATT_IS_COMPRESSED(attr)); - cmoid = GetCompressionOidFromCompressionId(TOAST_COMPRESS_METHOD(attr)); + /* get the compression method oid */ + cmoid = toast_get_compression_oid(attr); - /* Get compression method handler routines */ + /* get compression method handler routines */ cmroutine = GetCompressionRoutine(cmoid); - return cmroutine->cmdecompress(attr); + /* call the decompression routine */ + return cmroutine->cmdecompress(attr, + IsCustomeCompressionID(GetCompressionId(cmoid)) ? + TOAST_CUSTOM_COMPRESS_HDRSZ : TOAST_COMPRESS_HDRSZ); } @@ -471,23 +504,28 @@ static struct varlena * toast_decompress_datum_slice(struct varlena *attr, int32 slicelength) { CompressionRoutine *cmroutine; - Oid cmoid; + Oid cmoid; + int32 header_sz; Assert(VARATT_IS_COMPRESSED(attr)); - cmoid = GetCompressionOidFromCompressionId(TOAST_COMPRESS_METHOD(attr)); + cmoid = toast_get_compression_oid(attr); /* Get compression method handler routines */ cmroutine = GetCompressionRoutine(cmoid); + /* get the compression header size */ + header_sz = IsCustomeCompressionID(GetCompressionId(cmoid)) ? + TOAST_CUSTOM_COMPRESS_HDRSZ : TOAST_COMPRESS_HDRSZ; + /* * If the handler supports the slice decompression then only decompress * the slice otherwise decompress complete data. */ if (cmroutine->cmdecompress_slice) - return cmroutine->cmdecompress_slice(attr, slicelength); + return cmroutine->cmdecompress_slice(attr, header_sz, slicelength); else - return cmroutine->cmdecompress(attr); + return cmroutine->cmdecompress(attr, header_sz); } /* ---------- diff --git a/src/backend/access/common/toast_internals.c b/src/backend/access/common/toast_internals.c index 7c603dd19a..4bf383a43a 100644 --- a/src/backend/access/common/toast_internals.c +++ b/src/backend/access/common/toast_internals.c @@ -38,10 +38,14 @@ static bool toastid_valueid_exists(Oid toastrelid, Oid valueid); * -------- */ void -toast_set_compressed_datum_info(struct varlena *val, CompressionId cmid, int32 rawsize) +toast_set_compressed_datum_info(struct varlena *val, CompressionId cmid, + Oid cmoid, int32 rawsize) { TOAST_COMPRESS_SET_RAWSIZE(val, rawsize); TOAST_COMPRESS_SET_COMPRESSION_METHOD(val, cmid); + + if (IsCustomeCompressionID(cmid)) + TOAST_COMPRESS_SET_CMID(val, cmoid); } /* ---------- @@ -62,6 +66,7 @@ toast_compress_datum(Datum value, Oid cmoid) { struct varlena *tmp = NULL; int32 valsize; + CompressionId cmid; CompressionRoutine *cmroutine; Assert(!VARATT_IS_EXTERNAL(DatumGetPointer(value))); @@ -73,9 +78,12 @@ toast_compress_datum(Datum value, Oid cmoid) /* Get the compression routines for the compression method */ cmroutine = GetCompressionRoutine(cmoid); + cmid = GetCompressionId(cmoid); /* Call the actual compression function */ - tmp = cmroutine->cmcompress((const struct varlena *) value); + tmp = cmroutine->cmcompress((const struct varlena *) value, + IsCustomeCompressionID(cmid) ? + TOAST_CUSTOM_COMPRESS_HDRSZ : TOAST_COMPRESS_HDRSZ); if (!tmp) return PointerGetDatum(NULL); @@ -94,8 +102,7 @@ toast_compress_datum(Datum value, Oid cmoid) if (VARSIZE(tmp) < valsize - 2) { /* successful compression */ - toast_set_compressed_datum_info(tmp, GetCompressionId(cmoid), - valsize); + toast_set_compressed_datum_info(tmp, cmid, cmoid, valsize); return PointerGetDatum(tmp); } else diff --git a/src/backend/access/compression/compressionapi.c b/src/backend/access/compression/compressionapi.c index 4c82524143..6e0e7d48c8 100644 --- a/src/backend/access/compression/compressionapi.c +++ b/src/backend/access/compression/compressionapi.c @@ -131,6 +131,6 @@ GetCompressionId(Oid cmoid) case ZLIB_COMPRESSION_OID: return ZLIB_COMPRESSION_ID; default: - elog(ERROR, "Invalid compression method %d", cmoid); + return CUSTOME_COMPRESSION_ID; } } diff --git a/src/backend/access/compression/pglz.c b/src/backend/access/compression/pglz.c index 61dfe42004..931394f779 100644 --- a/src/backend/access/compression/pglz.c +++ b/src/backend/access/compression/pglz.c @@ -27,7 +27,7 @@ * compressed varlena, or NULL if compression fails. */ static struct varlena * -pglz_cmcompress(const struct varlena *value) +pglz_cmcompress(const struct varlena *value, int32 header_size) { int32 valsize, len; @@ -44,16 +44,16 @@ pglz_cmcompress(const struct varlena *value) return NULL; tmp = (struct varlena *) palloc(PGLZ_MAX_OUTPUT(valsize) + - TOAST_COMPRESS_HDRSZ); + header_size); len = pglz_compress(VARDATA_ANY(DatumGetPointer(value)), valsize, - TOAST_COMPRESS_RAWDATA(tmp), + (char *) tmp + header_size, NULL); if (len >= 0) { - SET_VARSIZE_COMPRESSED(tmp, len + TOAST_COMPRESS_HDRSZ); + SET_VARSIZE_COMPRESSED(tmp, len + header_size); return tmp; } @@ -68,7 +68,7 @@ pglz_cmcompress(const struct varlena *value) * Returns the decompressed varlena. */ static struct varlena * -pglz_cmdecompress(const struct varlena *value) +pglz_cmdecompress(const struct varlena *value, int32 header_size) { struct varlena *result; int32 rawsize; @@ -76,8 +76,8 @@ pglz_cmdecompress(const struct varlena *value) result = (struct varlena *) palloc(TOAST_COMPRESS_RAWSIZE(value) + VARHDRSZ); SET_VARSIZE(result, TOAST_COMPRESS_RAWSIZE(value) + VARHDRSZ); - rawsize = pglz_decompress(TOAST_COMPRESS_RAWDATA(value), - TOAST_COMPRESS_SIZE(value), + rawsize = pglz_decompress((char *) value + header_size, + VARSIZE(value) - header_size, VARDATA(result), TOAST_COMPRESS_RAWSIZE(value), true); @@ -95,7 +95,7 @@ pglz_cmdecompress(const struct varlena *value) * Decompresses part of the data. Returns the decompressed varlena. */ static struct varlena * -pglz_cmdecompress_slice(const struct varlena *value, +pglz_cmdecompress_slice(const struct varlena *value, int32 header_size, int32 slicelength) { struct varlena *result; @@ -103,8 +103,8 @@ pglz_cmdecompress_slice(const struct varlena *value, result = (struct varlena *) palloc(slicelength + VARHDRSZ); - rawsize = pglz_decompress(TOAST_COMPRESS_RAWDATA(value), - VARSIZE(value) - TOAST_COMPRESS_HDRSZ, + rawsize = pglz_decompress((char *) value + header_size, + VARSIZE(value) - header_size, VARDATA(result), slicelength, false); diff --git a/src/backend/access/compression/zlib.c b/src/backend/access/compression/zlib.c index 7cd106601d..eb31184aef 100644 --- a/src/backend/access/compression/zlib.c +++ b/src/backend/access/compression/zlib.c @@ -38,7 +38,7 @@ typedef struct * Returns the compressed varlena, or NULL if compression fails. */ static struct varlena * -zlib_cmcompress(const struct varlena *value) +zlib_cmcompress(const struct varlena *value, int32 header_size) { int32 valsize, len; @@ -58,11 +58,11 @@ zlib_cmcompress(const struct varlena *value) elog(ERROR, "could not initialize compression library: %s", zp->msg); valsize = VARSIZE_ANY_EXHDR(DatumGetPointer(value)); - tmp = (struct varlena *) palloc(valsize + TOAST_COMPRESS_HDRSZ); + tmp = (struct varlena *) palloc(valsize + header_size); zp->next_in = (void *) VARDATA_ANY(value); zp->avail_in = valsize; zp->avail_out = valsize; - zp->next_out = (void *) ((char *) tmp + TOAST_COMPRESS_HDRSZ); + zp->next_out = (void *) ((char *) tmp + header_size); do { @@ -80,7 +80,7 @@ zlib_cmcompress(const struct varlena *value) if (len > 0) { - SET_VARSIZE_COMPRESSED(tmp, len + TOAST_COMPRESS_HDRSZ); + SET_VARSIZE_COMPRESSED(tmp, len + header_size); return tmp; } @@ -94,7 +94,7 @@ zlib_cmcompress(const struct varlena *value) * Returns the decompressed varlena. */ static struct varlena * -zlib_cmdecompress(const struct varlena *value) +zlib_cmdecompress(const struct varlena *value, int32 header_size) { struct varlena *result; z_streamp zp; @@ -108,8 +108,8 @@ zlib_cmdecompress(const struct varlena *value) if (inflateInit(zp) != Z_OK) elog(ERROR, "could not initialize compression library: %s", zp->msg); - zp->next_in = (void *) ((char *) value + TOAST_COMPRESS_HDRSZ); - zp->avail_in = VARSIZE(value) - TOAST_COMPRESS_HDRSZ; + zp->next_in = (void *) ((char *) value + header_size); + zp->avail_in = VARSIZE(value) - header_size; zp->avail_out = VARRAWSIZE_4B_C(value); result = (struct varlena *) palloc(zp->avail_out + VARHDRSZ); diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c index c626161408..bf47b230b1 100644 --- a/src/backend/catalog/aclchk.c +++ b/src/backend/catalog/aclchk.c @@ -3410,6 +3410,7 @@ aclcheck_error(AclResult aclerr, ObjectType objtype, case OBJECT_AMPROC: case OBJECT_ATTRIBUTE: case OBJECT_CAST: + case OBJECT_COMPRESSION_METHOD: case OBJECT_DEFAULT: case OBJECT_DEFACL: case OBJECT_DOMCONSTRAINT: @@ -3549,6 +3550,7 @@ aclcheck_error(AclResult aclerr, ObjectType objtype, case OBJECT_AMPROC: case OBJECT_ATTRIBUTE: case OBJECT_CAST: + case OBJECT_COMPRESSION_METHOD: case OBJECT_DEFAULT: case OBJECT_DEFACL: case OBJECT_DOMCONSTRAINT: diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c index 6e86c66456..6db2cee3ca 100644 --- a/src/backend/catalog/dependency.c +++ b/src/backend/catalog/dependency.c @@ -1487,6 +1487,7 @@ doDeletion(const ObjectAddress *object, int flags) case OCLASS_AM: case OCLASS_AMOP: case OCLASS_AMPROC: + case OCLASS_COMPRESSION: case OCLASS_SCHEMA: case OCLASS_TSPARSER: case OCLASS_TSDICT: @@ -1508,7 +1509,6 @@ doDeletion(const ObjectAddress *object, int flags) case OCLASS_DATABASE: case OCLASS_TBLSPACE: case OCLASS_SUBSCRIPTION: - case OCLASS_COMPRESSION: elog(ERROR, "global objects cannot be deleted by doDeletion"); break; diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c index ed40d78d8d..1f98f28f6b 100644 --- a/src/backend/catalog/objectaddress.c +++ b/src/backend/catalog/objectaddress.c @@ -190,6 +190,20 @@ static const ObjectPropertyType ObjectProperty[] = OBJECT_COLLATION, true }, + { + "compression method", + CompressionRelationId, + CompressionIndexId, + CMOID, + CMNAME, + Anum_pg_compression_oid, + Anum_pg_compression_cmname, + InvalidAttrNumber, + InvalidAttrNumber, + InvalidAttrNumber, + -1, + true + }, { "constraint", ConstraintRelationId, @@ -1010,6 +1024,7 @@ get_object_address(ObjectType objtype, Node *object, case OBJECT_FOREIGN_SERVER: case OBJECT_EVENT_TRIGGER: case OBJECT_ACCESS_METHOD: + case OBJECT_COMPRESSION_METHOD: case OBJECT_PUBLICATION: case OBJECT_SUBSCRIPTION: address = get_object_address_unqualified(objtype, @@ -1261,6 +1276,11 @@ get_object_address_unqualified(ObjectType objtype, address.objectId = get_am_oid(name, missing_ok); address.objectSubId = 0; break; + case OBJECT_COMPRESSION_METHOD: + address.classId = CompressionRelationId; + address.objectId = get_cm_oid(name, missing_ok); + address.objectSubId = 0; + break; case OBJECT_DATABASE: address.classId = DatabaseRelationId; address.objectId = get_database_oid(name, missing_ok); @@ -2272,6 +2292,7 @@ pg_get_object_address(PG_FUNCTION_ARGS) objnode = (Node *) name; break; case OBJECT_ACCESS_METHOD: + case OBJECT_COMPRESSION_METHOD: case OBJECT_DATABASE: case OBJECT_EVENT_TRIGGER: case OBJECT_EXTENSION: @@ -2565,6 +2586,7 @@ 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, diff --git a/src/backend/commands/compressioncmds.c b/src/backend/commands/compressioncmds.c index 15fb3bc40a..719530515b 100644 --- a/src/backend/commands/compressioncmds.c +++ b/src/backend/commands/compressioncmds.c @@ -20,11 +20,114 @@ #include "catalog/catalog.h" #include "catalog/dependency.h" #include "catalog/indexing.h" +#include "catalog/objectaccess.h" #include "catalog/pg_attribute.h" #include "catalog/pg_depend.h" +#include "catalog/pg_proc.h" #include "commands/defrem.h" +#include "miscadmin.h" #include "nodes/parsenodes.h" +#include "parser/parse_func.h" #include "utils/fmgroids.h" +#include "utils/fmgrprotos.h" +#include "utils/syscache.h" + +/* + * CreateCompressionMethod + * Registers a new compression method. + */ +ObjectAddress +CreateCompressionMethod(CreateCmStmt *stmt) +{ + Relation rel; + ObjectAddress myself; + ObjectAddress referenced; + Oid cmoid; + Oid cmhandler; + bool nulls[Natts_pg_compression]; + Datum values[Natts_pg_compression]; + HeapTuple tup; + Oid funcargtypes[1] = {INTERNALOID}; + + rel = table_open(CompressionRelationId, RowExclusiveLock); + + /* Must be super user */ + if (!superuser()) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("permission denied to create compression method \"%s\"", + stmt->cmname), + errhint("Must be superuser to create an access method."))); + + /* Check if name is used */ + cmoid = GetSysCacheOid1(CMNAME, Anum_pg_compression_oid, + CStringGetDatum(stmt->cmname)); + if (OidIsValid(cmoid)) + { + ereport(ERROR, + (errcode(ERRCODE_DUPLICATE_OBJECT), + errmsg("compression method \"%s\" already exists", + stmt->cmname))); + } + + if (stmt->handler_name == NIL) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_FUNCTION), + errmsg("handler function is not specified"))); + + /* Get the handler function oid */ + cmhandler = LookupFuncName(stmt->handler_name, 1, funcargtypes, false); + + /* + * Insert tuple into pg_compression. + */ + memset(values, 0, sizeof(values)); + memset(nulls, false, sizeof(nulls)); + + cmoid = GetNewOidWithIndex(rel, AmOidIndexId, Anum_pg_compression_oid); + values[Anum_pg_compression_oid - 1] = ObjectIdGetDatum(cmoid); + values[Anum_pg_compression_cmname - 1] = + DirectFunctionCall1(namein, CStringGetDatum(stmt->cmname)); + values[Anum_pg_compression_cmhandler - 1] = ObjectIdGetDatum(cmhandler); + + tup = heap_form_tuple(RelationGetDescr(rel), values, nulls); + + CatalogTupleInsert(rel, tup); + heap_freetuple(tup); + + myself.classId = CompressionRelationId; + myself.objectId = cmoid; + myself.objectSubId = 0; + + /* Record dependency on handler function */ + referenced.classId = ProcedureRelationId; + referenced.objectId = cmhandler; + referenced.objectSubId = 0; + + recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL); + + recordDependencyOnCurrentExtension(&myself, false); + + InvokeObjectPostCreateHook(CompressionRelationId, cmoid, 0); + + table_close(rel, RowExclusiveLock); + + return myself; +} + +Oid +get_cm_oid(const char *cmname, bool missing_ok) +{ + Oid cmoid; + + cmoid = GetCompressionOid(cmname); + if (!OidIsValid(cmoid) && !missing_ok) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("compression type \"%s\" not recognized", cmname))); + + return cmoid; +} /* * get list of all supported compression methods for the given attribute. diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c index 26fe640e8e..57bffc4691 100644 --- a/src/backend/commands/event_trigger.c +++ b/src/backend/commands/event_trigger.c @@ -953,6 +953,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: @@ -2106,6 +2107,7 @@ stringify_grant_objtype(ObjectType objtype) case OBJECT_ATTRIBUTE: case OBJECT_CAST: case OBJECT_COLLATION: + case OBJECT_COMPRESSION_METHOD: case OBJECT_CONVERSION: case OBJECT_DEFAULT: case OBJECT_DEFACL: @@ -2188,6 +2190,7 @@ stringify_adefprivs_objtype(ObjectType objtype) case OBJECT_ATTRIBUTE: case OBJECT_CAST: case OBJECT_COLLATION: + case OBJECT_COMPRESSION_METHOD: case OBJECT_CONVERSION: case OBJECT_DEFAULT: case OBJECT_DEFACL: diff --git a/src/backend/commands/seclabel.c b/src/backend/commands/seclabel.c index ee036e9087..083ac78e11 100644 --- a/src/backend/commands/seclabel.c +++ b/src/backend/commands/seclabel.c @@ -67,6 +67,7 @@ SecLabelSupportsObjectType(ObjectType objtype) case OBJECT_ATTRIBUTE: case OBJECT_CAST: case OBJECT_COLLATION: + case OBJECT_COMPRESSION_METHOD: case OBJECT_CONVERSION: case OBJECT_DEFAULT: case OBJECT_DEFACL: diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index 0d0d29f9ce..4aad3829df 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -2109,8 +2109,7 @@ CheckTargetCMAndDecompress(TupleTableSlot *slot, TupleDesc targetTupDesc) * Get the compression method stored in the toast header and * compare with the compression method of the target. */ - cmoid = GetCompressionOidFromCompressionId( - TOAST_COMPRESS_METHOD(new_value)); + cmoid = toast_get_compression_oid(new_value); if (!IsCompressionSupported(&targetTupDesc->attrs[i], cmoid)) { new_value = detoast_attr(new_value); diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 5095c0e09c..018de03016 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -4319,6 +4319,17 @@ _copyCreateAmStmt(const CreateAmStmt *from) return newnode; } +static CreateCmStmt * +_copyCreateCmStmt(const CreateCmStmt *from) +{ + CreateCmStmt *newnode = makeNode(CreateCmStmt); + + COPY_STRING_FIELD(cmname); + COPY_NODE_FIELD(handler_name); + + return newnode; +} + static CreateTrigStmt * _copyCreateTrigStmt(const CreateTrigStmt *from) { @@ -5484,6 +5495,9 @@ copyObjectImpl(const void *from) case T_CreateAmStmt: retval = _copyCreateAmStmt(from); break; + case T_CreateCmStmt: + retval = _copyCreateCmStmt(from); + break; case T_CreateTrigStmt: retval = _copyCreateTrigStmt(from); break; diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c index 35efa064e5..5057cf7529 100644 --- a/src/backend/nodes/equalfuncs.c +++ b/src/backend/nodes/equalfuncs.c @@ -2016,6 +2016,15 @@ _equalCreateAmStmt(const CreateAmStmt *a, const CreateAmStmt *b) return true; } +static bool +_equalCreateCmStmt(const CreateCmStmt *a, const CreateCmStmt *b) +{ + COMPARE_STRING_FIELD(cmname); + COMPARE_NODE_FIELD(handler_name); + + return true; +} + static bool _equalCreateTrigStmt(const CreateTrigStmt *a, const CreateTrigStmt *b) { @@ -3536,6 +3545,9 @@ equal(const void *a, const void *b) case T_CreateAmStmt: retval = _equalCreateAmStmt(a, b); break; + case T_CreateCmStmt: + retval = _equalCreateCmStmt(a, b); + break; case T_CreateTrigStmt: retval = _equalCreateTrigStmt(a, b); break; diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 47d18a79ba..1de3e2354b 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -289,7 +289,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); DeallocateStmt PrepareStmt ExecuteStmt DropOwnedStmt ReassignOwnedStmt AlterTSConfigurationStmt AlterTSDictionaryStmt - CreateMatViewStmt RefreshMatViewStmt CreateAmStmt + CreateMatViewStmt RefreshMatViewStmt CreateAmStmt CreateCmStmt CreatePublicationStmt AlterPublicationStmt CreateSubscriptionStmt AlterSubscriptionStmt DropSubscriptionStmt @@ -882,6 +882,7 @@ stmt : | CreateAsStmt | CreateAssertionStmt | CreateCastStmt + | CreateCmStmt | CreateConversionStmt | CreateDomainStmt | CreateExtensionStmt @@ -5248,6 +5249,22 @@ am_type: | TABLE { $$ = AMTYPE_TABLE; } ; +/***************************************************************************** + * + * QUERY: + * CREATE COMPRESSION METHOD name HANDLER handler_name + * + *****************************************************************************/ + +CreateCmStmt: CREATE COMPRESSION METHOD name HANDLER handler_name + { + CreateCmStmt *n = makeNode(CreateCmStmt); + n->cmname = $4; + n->handler_name = $6; + $$ = (Node *) n; + } + ; + /***************************************************************************** * * QUERIES : @@ -6250,6 +6267,7 @@ object_type_name: drop_type_name: ACCESS METHOD { $$ = OBJECT_ACCESS_METHOD; } + | COMPRESSION METHOD { $$ = OBJECT_COMPRESSION_METHOD; } | EVENT TRIGGER { $$ = OBJECT_EVENT_TRIGGER; } | EXTENSION { $$ = OBJECT_EXTENSION; } | FOREIGN DATA_P WRAPPER { $$ = OBJECT_FDW; } diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index 9a35147b26..7ebbf51778 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -168,6 +168,7 @@ ClassifyUtilityCommandAsReadOnly(Node *parsetree) case T_CompositeTypeStmt: case T_CreateAmStmt: case T_CreateCastStmt: + case T_CreateCmStmt: case T_CreateConversionStmt: case T_CreateDomainStmt: case T_CreateEnumStmt: @@ -1805,6 +1806,10 @@ ProcessUtilitySlow(ParseState *pstate, address = CreateAccessMethod((CreateAmStmt *) parsetree); break; + case T_CreateCmStmt: + address = CreateCompressionMethod((CreateCmStmt *) parsetree); + break; + case T_CreatePublicationStmt: address = CreatePublication((CreatePublicationStmt *) parsetree); break; @@ -2566,6 +2571,9 @@ CreateCommandTag(Node *parsetree) case OBJECT_ACCESS_METHOD: tag = CMDTAG_DROP_ACCESS_METHOD; break; + case OBJECT_COMPRESSION_METHOD: + tag = CMDTAG_DROP_COMPRESSION_METHOD; + break; case OBJECT_PUBLICATION: tag = CMDTAG_DROP_PUBLICATION; break; @@ -2973,6 +2981,10 @@ CreateCommandTag(Node *parsetree) tag = CMDTAG_CREATE_ACCESS_METHOD; break; + case T_CreateCmStmt: + tag = CMDTAG_CREATE_COMPRESSION_METHOD; + break; + case T_CreatePublicationStmt: tag = CMDTAG_CREATE_PUBLICATION; break; @@ -3581,6 +3593,10 @@ GetCommandLogLevel(Node *parsetree) lev = LOGSTMT_DDL; break; + case T_CreateCmStmt: + lev = LOGSTMT_DDL; + break; + case T_CreatePublicationStmt: lev = LOGSTMT_DDL; break; diff --git a/src/include/access/compressionapi.h b/src/include/access/compressionapi.h index f905f3ecaf..dfc0c8b64f 100644 --- a/src/include/access/compressionapi.h +++ b/src/include/access/compressionapi.h @@ -26,18 +26,22 @@ */ typedef enum CompressionId { - PGLZ_COMPRESSION_ID, - ZLIB_COMPRESSION_ID + PGLZ_COMPRESSION_ID = 0, + ZLIB_COMPRESSION_ID = 1, + CUSTOME_COMPRESSION_ID = 3 } CompressionId; - -#define DefaultCompressionOid PGLZ_COMPRESSION_OID +#define DefaultCompressionOid PGLZ_COMPRESSION_OID +#define IsCustomeCompressionID(cmid) ((cmid) == CUSTOME_COMPRESSION_ID) typedef struct CompressionRoutine CompressionRoutine; -typedef struct varlena *(*cmcompress_function) (const struct varlena *value); +typedef struct varlena *(*cmcompress_function) (const struct varlena *value, + int32 toast_header_size); typedef struct varlena *(*cmdecompress_slice_function) - (const struct varlena *value, int32 slicelength); + (const struct varlena *value, + int32 toast_header_size, + int32 slicelength); /* * API struct for a compression routine. diff --git a/src/include/access/detoast.h b/src/include/access/detoast.h index 86bad7e78c..3c280ce953 100644 --- a/src/include/access/detoast.h +++ b/src/include/access/detoast.h @@ -89,4 +89,12 @@ extern Size toast_raw_datum_size(Datum value); */ extern Size toast_datum_size(Datum value); +/* ---------- + * toast_get_compression_oid - + * + * Return the Oid of the stored compression method + * ---------- + */ +extern Oid toast_get_compression_oid(struct varlena *attr); + #endif /* DETOAST_H */ diff --git a/src/include/access/toast_internals.h b/src/include/access/toast_internals.h index 8ef477cfe0..48ca172eae 100644 --- a/src/include/access/toast_internals.h +++ b/src/include/access/toast_internals.h @@ -27,6 +27,17 @@ typedef struct toast_compress_header * 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; /* 2 bits for compression method + rawsize */ + Oid cmoid; /* Oid from pg_compression */ +} toast_compress_header_custom; + #define RAWSIZEMASK (0x3FFFFFFFU) /* @@ -38,6 +49,7 @@ typedef struct toast_compress_header * two highest bits. */ #define TOAST_COMPRESS_HDRSZ ((int32) sizeof(toast_compress_header)) +#define TOAST_CUSTOM_COMPRESS_HDRSZ ((int32) sizeof(toast_compress_header_custom)) #define TOAST_COMPRESS_RAWSIZE(ptr) (((toast_compress_header *) (ptr))->info & RAWSIZEMASK) #define TOAST_COMPRESS_METHOD(ptr) (((toast_compress_header *) (ptr))->info >> 30) #define TOAST_COMPRESS_SIZE(ptr) ((int32) VARSIZE_ANY(ptr) - TOAST_COMPRESS_HDRSZ) @@ -52,6 +64,9 @@ do { \ #define TOAST_COMPRESS_SET_COMPRESSION_METHOD(ptr, cm_method) \ ((toast_compress_header *) (ptr))->info |= ((cm_method) << 30); +#define TOAST_COMPRESS_SET_CMID(ptr, oid) \ + (((toast_compress_header_custom *) (ptr))->cmoid = (oid)) + extern Datum toast_compress_datum(Datum value, Oid cmoid); extern Oid toast_get_valid_index(Oid toastoid, LOCKMODE lock); @@ -72,7 +87,9 @@ extern void init_toast_snapshot(Snapshot toast_snapshot); * * Save metadata in compressed datum */ -extern void toast_set_compressed_datum_info(struct varlena *val, CompressionId cmid, +extern void toast_set_compressed_datum_info(struct varlena *val, + CompressionId cmid, + Oid cmoid, int32 rawsize); #endif /* TOAST_INTERNALS_H */ diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h index 936b072c76..8952b2b70d 100644 --- a/src/include/commands/defrem.h +++ b/src/include/commands/defrem.h @@ -146,6 +146,8 @@ extern Oid get_am_oid(const char *amname, bool missing_ok); extern char *get_am_name(Oid amOid); /* commands/compressioncmds.c */ +extern ObjectAddress CreateCompressionMethod(CreateCmStmt *stmt); +extern Oid get_cm_oid(const char *cmname, bool missing_ok); extern Oid GetAttributeCompression(Form_pg_attribute att, ColumnCompression *compression, bool *need_rewrite); diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h index 65ef987d67..ae297c9116 100644 --- a/src/include/nodes/nodes.h +++ b/src/include/nodes/nodes.h @@ -415,6 +415,7 @@ typedef enum NodeTag T_AlterPolicyStmt, T_CreateTransformStmt, T_CreateAmStmt, + T_CreateCmStmt, T_CreatePublicationStmt, T_AlterPublicationStmt, T_CreateSubscriptionStmt, @@ -493,7 +494,7 @@ typedef enum NodeTag T_TimeLineHistoryCmd, T_SQLCmd, - T_CompressionRoutine, /* in access/compressionapi.h */ + T_CompressionRoutine, /* in access/compressionapi.h */ /* * TAGS FOR RANDOM OTHER STUFF * diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index e413914979..0f4aee5a6d 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -1716,6 +1716,7 @@ typedef enum ObjectType OBJECT_CAST, OBJECT_COLUMN, OBJECT_COLLATION, + OBJECT_COMPRESSION_METHOD, OBJECT_CONVERSION, OBJECT_DATABASE, OBJECT_DEFAULT, @@ -2438,6 +2439,17 @@ typedef struct CreateAmStmt char amtype; /* type of access method */ } CreateAmStmt; +/*---------------------- + * Create COMPRESSION METHOD Statement + *---------------------- + */ +typedef struct CreateCmStmt +{ + NodeTag type; + char *cmname; /* compression method name */ + List *handler_name; /* handler function name */ +} CreateCmStmt; + /* ---------------------- * Create TRIGGER Statement * ---------------------- diff --git a/src/include/postgres.h b/src/include/postgres.h index 04c94f489d..bb4ca6a383 100644 --- a/src/include/postgres.h +++ b/src/include/postgres.h @@ -149,6 +149,14 @@ typedef union * 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 method */ + char va_data[FLEXIBLE_ARRAY_MEMBER]; /* Compressed data */ + } va_custom_compressed; } varattrib_4b; typedef struct diff --git a/src/include/tcop/cmdtaglist.h b/src/include/tcop/cmdtaglist.h index be94852bbd..a9bc1c4bae 100644 --- a/src/include/tcop/cmdtaglist.h +++ b/src/include/tcop/cmdtaglist.h @@ -83,6 +83,7 @@ PG_CMDTAG(CMDTAG_COMMIT_PREPARED, "COMMIT PREPARED", false, false, false) PG_CMDTAG(CMDTAG_COPY, "COPY", false, false, true) PG_CMDTAG(CMDTAG_COPY_FROM, "COPY FROM", false, false, false) PG_CMDTAG(CMDTAG_CREATE_ACCESS_METHOD, "CREATE ACCESS METHOD", true, false, false) +PG_CMDTAG(CMDTAG_CREATE_COMPRESSION_METHOD, "CREATE COMPRESSION METHOD", true, false, false) PG_CMDTAG(CMDTAG_CREATE_AGGREGATE, "CREATE AGGREGATE", true, false, false) PG_CMDTAG(CMDTAG_CREATE_CAST, "CREATE CAST", true, false, false) PG_CMDTAG(CMDTAG_CREATE_COLLATION, "CREATE COLLATION", true, false, false) @@ -138,6 +139,7 @@ PG_CMDTAG(CMDTAG_DROP_ACCESS_METHOD, "DROP ACCESS METHOD", true, false, false) PG_CMDTAG(CMDTAG_DROP_AGGREGATE, "DROP AGGREGATE", true, false, false) PG_CMDTAG(CMDTAG_DROP_CAST, "DROP CAST", true, false, false) PG_CMDTAG(CMDTAG_DROP_COLLATION, "DROP COLLATION", true, false, false) +PG_CMDTAG(CMDTAG_DROP_COMPRESSION_METHOD, "DROP COMPRESSIOn METHOD", true, false, false) PG_CMDTAG(CMDTAG_DROP_CONSTRAINT, "DROP CONSTRAINT", true, false, false) PG_CMDTAG(CMDTAG_DROP_CONVERSION, "DROP CONVERSION", true, false, false) PG_CMDTAG(CMDTAG_DROP_DATABASE, "DROP DATABASE", false, false, false) diff --git a/src/test/regress/expected/create_cm.out b/src/test/regress/expected/create_cm.out index 33c7132353..bf89651b4f 100644 --- a/src/test/regress/expected/create_cm.out +++ b/src/test/regress/expected/create_cm.out @@ -107,4 +107,22 @@ INSERT INTO cmmove2 VALUES (repeat('1234567890',1004)); --------+------+-----------+----------+---------+----------+-------------+--------------+------------- f1 | text | | | | extended | zlib | | +-- create compression method +CREATE COMPRESSION METHOD pglz2 HANDLER pglzhandler; +ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION pglz2; +INSERT INTO cmmove2 VALUES (repeat('1234567890',1004)); +SELECT length(f1) FROM cmmove2; + length +-------- + 10000 + 10040 + 10040 +(3 rows) + +\d+ cmmove2 + Table "public.cmmove2" + Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description +--------+------+-----------+----------+---------+----------+-------------+--------------+------------- + f1 | text | | | | extended | pglz2 | | + DROP TABLE cmmove1, cmmove2, cmmove3, zlibtest; diff --git a/src/test/regress/sql/create_cm.sql b/src/test/regress/sql/create_cm.sql index abf34eede0..273dcb83b9 100644 --- a/src/test/regress/sql/create_cm.sql +++ b/src/test/regress/sql/create_cm.sql @@ -52,4 +52,13 @@ ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION pglz; ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION zlib PRESERVE (pglz); INSERT INTO cmmove2 VALUES (repeat('1234567890',1004)); \d+ cmmove2 + +-- create compression method +CREATE COMPRESSION METHOD pglz2 HANDLER pglzhandler; +ALTER TABLE cmmove2 ALTER COLUMN f1 SET COMPRESSION pglz2; +INSERT INTO cmmove2 VALUES (repeat('1234567890',1004)); +SELECT length(f1) FROM cmmove2; +\d+ cmmove2 + + DROP TABLE cmmove1, cmmove2, cmmove3, zlibtest; -- 2.23.0