public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v24 7/7] Add support for PRESERVE 2+ messages / 2 participants [nested] [flat]
* [PATCH v24 7/7] Add support for PRESERVE @ 2021-02-10 06:19 Dilip Kumar <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Dilip Kumar @ 2021-02-10 06:19 UTC (permalink / raw) Now the compression method can be changed without forcing a table rewrite, by including the old method in the PRESERVE list. �For supporting this the column will maintain the dependency with all the supported compression methods. �So whenever the compression method is altered the dependency is added with the new compression method and the dependency is removed for all the old compression methods which are not given in the preserve list. �If PRESERVE ALL is given then all the dependency is maintained. Dilip Kumar based on the patches from Ildus Kurbangaliev. Design input from Tomas Vondra and Robert Haas Reviewed by Robert Haas, Tomas Vondra, Alexander Korotkov and Justin Pryzby --- doc/src/sgml/ref/alter_table.sgml | 10 +- src/backend/catalog/pg_depend.c | 7 + src/backend/commands/Makefile | 1 + src/backend/commands/compressioncmds.c | 300 ++++++++++++++++++++ src/backend/commands/tablecmds.c | 126 ++++---- src/backend/executor/nodeModifyTable.c | 12 +- src/backend/nodes/copyfuncs.c | 17 +- src/backend/nodes/equalfuncs.c | 15 +- src/backend/nodes/outfuncs.c | 15 +- src/backend/parser/gram.y | 52 +++- src/backend/parser/parse_utilcmd.c | 2 +- src/bin/pg_dump/pg_dump.c | 101 +++++++ src/bin/pg_dump/pg_dump.h | 15 +- src/bin/psql/tab-complete.c | 7 + src/include/commands/defrem.h | 7 + src/include/commands/tablecmds.h | 2 + src/include/nodes/nodes.h | 1 + src/include/nodes/parsenodes.h | 16 +- src/test/regress/expected/compression.out | 37 ++- src/test/regress/expected/compression_1.out | 39 ++- src/test/regress/expected/create_index.out | 56 ++-- src/test/regress/sql/compression.sql | 9 + 22 files changed, 739 insertions(+), 108 deletions(-) create mode 100644 src/backend/commands/compressioncmds.c diff --git a/doc/src/sgml/ref/alter_table.sgml b/doc/src/sgml/ref/alter_table.sgml index 0bd0c1a503..c9f443a59c 100644 --- a/doc/src/sgml/ref/alter_table.sgml +++ b/doc/src/sgml/ref/alter_table.sgml @@ -54,7 +54,7 @@ ALTER TABLE [ IF EXISTS ] <replaceable class="parameter">name</replaceable> ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET ( <replaceable class="parameter">attribute_option</replaceable> = <replaceable class="parameter">value</replaceable> [, ... ] ) ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> RESET ( <replaceable class="parameter">attribute_option</replaceable> [, ... ] ) ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET STORAGE { PLAIN | EXTERNAL | EXTENDED | MAIN } - ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET COMPRESSION <replaceable class="parameter">compression_method</replaceable> + ALTER [ COLUMN ] <replaceable class="parameter">column_name</replaceable> SET COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) | PRESERVE ALL ] ADD <replaceable class="parameter">table_constraint</replaceable> [ NOT VALID ] ADD <replaceable class="parameter">table_constraint_using_index</replaceable> ALTER CONSTRAINT <replaceable class="parameter">constraint_name</replaceable> [ DEFERRABLE | NOT DEFERRABLE ] [ INITIALLY DEFERRED | INITIALLY IMMEDIATE ] @@ -387,7 +387,7 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM <varlistentry> <term> - <literal>SET COMPRESSION <replaceable class="parameter">compression_method</replaceable></literal> + <literal>SET COMPRESSION <replaceable class="parameter">compression_method</replaceable> [ PRESERVE (<replaceable class="parameter">compression_preserve_list</replaceable>) | PRESERVE ALL ]</literal> </term> <listitem> <para> @@ -395,6 +395,12 @@ WITH ( MODULUS <replaceable class="parameter">numeric_literal</replaceable>, REM methods are <literal>pglz</literal> and <literal>lz4</literal>. <literal>lz4</literal> is available only if <literal>--with-lz4</literal> was used when building <productname>PostgreSQL</productname>. + The <literal>PRESERVE</literal> list contains a list of compression + methods used in the column and determines which of them may be kept. + Without <literal>PRESERVE</literal> or if any of the pre-existing + compression methods are not preserved, the table will be rewritten. If + <literal>PRESERVE ALL</literal> is specified, then all of the existing + methods will be preserved and the table will not be rewritten. </para> </listitem> </varlistentry> diff --git a/src/backend/catalog/pg_depend.c b/src/backend/catalog/pg_depend.c index 63da24322d..dd376484b7 100644 --- a/src/backend/catalog/pg_depend.c +++ b/src/backend/catalog/pg_depend.c @@ -17,6 +17,7 @@ #include "access/genam.h" #include "access/htup_details.h" #include "access/table.h" +#include "catalog/pg_am.h" #include "catalog/dependency.h" #include "catalog/indexing.h" #include "catalog/pg_collation.h" @@ -125,6 +126,12 @@ recordMultipleDependencies(const ObjectAddress *depender, if (referenced->objectId == DEFAULT_COLLATION_OID) ignore_systempin = true; } + /* + * Record the dependency on compression access method for handling + * preserve. + */ + if (referenced->classId == AccessMethodRelationId) + ignore_systempin = true; } else Assert(!version); diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile index e8504f0ae4..a7395ad77d 100644 --- a/src/backend/commands/Makefile +++ b/src/backend/commands/Makefile @@ -21,6 +21,7 @@ OBJS = \ cluster.o \ collationcmds.o \ comment.o \ + compressioncmds.o \ constraint.o \ conversioncmds.o \ copy.o \ diff --git a/src/backend/commands/compressioncmds.c b/src/backend/commands/compressioncmds.c new file mode 100644 index 0000000000..fd6db24e7f --- /dev/null +++ b/src/backend/commands/compressioncmds.c @@ -0,0 +1,300 @@ +/*------------------------------------------------------------------------- + * + * compressioncmds.c + * Routines for SQL commands for attribute compression methods + * + * Portions Copyright (c) 1996-2017, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * + * IDENTIFICATION + * src/backend/commands/compressioncmds.c + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/compressamapi.h" +#include "access/heapam.h" +#include "access/htup_details.h" +#include "access/reloptions.h" +#include "catalog/catalog.h" +#include "catalog/dependency.h" +#include "catalog/indexing.h" +#include "catalog/pg_attribute.h" +#include "catalog/pg_depend.h" +#include "commands/defrem.h" +#include "commands/tablecmds.h" +#include "miscadmin.h" +#include "nodes/parsenodes.h" +#include "utils/builtins.h" +#include "utils/fmgroids.h" +#include "utils/lsyscache.h" + +/* + * Get list of all supported compression methods for the given attribute. + * + * We maintain dependency of the attribute on the pg_am row for the current + * compression AM and all the preserved compression AM. So scan pg_depend and + * find the column dependency on the pg_am. Collect the list of access method + * oids on which this attribute has a dependency. + */ +static List * +lookup_attribute_compression(Oid attrelid, AttrNumber attnum, List *oldcmoids) +{ + LOCKMODE lock = AccessShareLock; + HeapTuple tup; + Relation rel; + SysScanDesc scan; + ScanKeyData key[3]; + List *cmoids = NIL; + + rel = table_open(DependRelationId, lock); + + ScanKeyInit(&key[0], + Anum_pg_depend_classid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(RelationRelationId)); + ScanKeyInit(&key[1], + Anum_pg_depend_objid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(attrelid)); + ScanKeyInit(&key[2], + Anum_pg_depend_objsubid, + BTEqualStrategyNumber, F_INT4EQ, + Int32GetDatum((int32) attnum)); + + scan = systable_beginscan(rel, DependDependerIndexId, true, + NULL, 3, key); + + while (HeapTupleIsValid(tup = systable_getnext(scan))) + { + Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tup); + + if (depform->refclassid == AccessMethodRelationId) + cmoids = list_append_unique_oid(cmoids, depform->refobjid); + } + + systable_endscan(scan); + table_close(rel, lock); + + return cmoids; +} + +/* + * Remove the attribute dependency on the old compression methods + * + * Scan the pg_depend and search the attribute dependency on the pg_am. Remove + * dependency on previous am which is not preserved. The list of non-preserved + * AMs is given in cmoids. + */ +static void +remove_old_dependencies(Oid attrelid, AttrNumber attnum, List *cmoids) +{ + LOCKMODE lock = AccessShareLock; + HeapTuple tup; + Relation rel; + SysScanDesc scan; + ScanKeyData key[3]; + + rel = table_open(DependRelationId, lock); + + ScanKeyInit(&key[0], + Anum_pg_depend_classid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(RelationRelationId)); + ScanKeyInit(&key[1], + Anum_pg_depend_objid, + BTEqualStrategyNumber, F_OIDEQ, + ObjectIdGetDatum(attrelid)); + ScanKeyInit(&key[2], + Anum_pg_depend_objsubid, + BTEqualStrategyNumber, F_INT4EQ, + Int32GetDatum((int32) attnum)); + + scan = systable_beginscan(rel, DependDependerIndexId, true, + NULL, 3, key); + + while (HeapTupleIsValid(tup = systable_getnext(scan))) + { + Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tup); + + if (list_member_oid(cmoids, depform->refobjid)) + { + Assert(depform->refclassid == AccessMethodRelationId); + CatalogTupleDelete(rel, &tup->t_self); + } + } + + systable_endscan(scan); + table_close(rel, lock); +} + +/* + * Check whether the given compression method oid is supported by + * the target attribute. + */ +bool +IsCompressionSupported(Form_pg_attribute att, Oid cmoid) +{ + List *cmoids = NIL; + + /* Check whether it is same as the current compression oid */ + if (cmoid == att->attcompression) + return true; + + /* Check the oid in all preserved compresion methods */ + cmoids = lookup_attribute_compression(att->attrelid, att->attnum, NULL); + if (list_member_oid(cmoids, cmoid)) + return true; + else + return false; +} + +/* + * In binary upgrade mode add the dependencies for all the preserved compression + * method. + */ +static void +BinaryUpgradeAddPreserve(Form_pg_attribute att, List *preserve) +{ + ListCell *cell; + + foreach(cell, preserve) + { + char *cmname_p = strVal(lfirst(cell)); + Oid cmoid_p = get_compression_am_oid(cmname_p, false); + + add_column_compression_dependency(att->attrelid, att->attnum, cmoid_p); + } +} + +/* + * Get the compression method oid based on the compression method name. When + * compression is not specified returns default attribute compression. It is + * possible case for CREATE TABLE and ADD COLUMN commands where COMPRESSION + * syntax is optional. + * + * For ALTER command, check all the supported compression methods for the + * attribute and if the preserve list is not passed or some of the old + * compression methods are not given in the preserved list then delete + * dependency from the old compression methods and force the table rewrite. + */ +Oid +GetAttributeCompression(Form_pg_attribute att, ColumnCompression *compression, + bool *need_rewrite) +{ + Oid cmoid; + char typstorage = get_typstorage(att->atttypid); + ListCell *cell; + + /* + * No compression for the plain/external storage, refer comments atop + * attcompression parameter in pg_attribute.h + */ + if (!IsStorageCompressible(typstorage)) + { + if (compression == NULL) + return InvalidOid; + + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("column data type %s does not support compression", + format_type_be(att->atttypid)))); + } + + /* fallback to default compression if it's not specified */ + if (compression == NULL) + return GetDefaultToastCompression(); + + cmoid = get_compression_am_oid(compression->cmname, false); + +#ifndef HAVE_LIBLZ4 + if (cmoid == LZ4_COMPRESSION_AM_OID) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("not built with lz4 support"))); +#endif + + /* + * Determine if the column needs rewrite or not. Rewrite conditions: SET + * COMPRESSION without PRESERVE - SET COMPRESSION with PRESERVE but not + * with full list of previous access methods. + */ + if (need_rewrite != NULL) + { + List *previous_cmoids = NIL; + + *need_rewrite = false; + + /* + * In binary upgrade mode, just create a dependency on all preserved + * methods. + */ + if (IsBinaryUpgrade) + { + BinaryUpgradeAddPreserve(att, compression->preserve); + return cmoid; + } + + /* If we have preserved all then rewrite is not required */ + if (compression->preserve_all) + return cmoid; + + previous_cmoids = + lookup_attribute_compression(att->attrelid, att->attnum, NULL); + + foreach(cell, compression->preserve) + { + char *cmname_p = strVal(lfirst(cell)); + Oid cmoid_p = get_compression_am_oid(cmname_p, false); + + if (!list_member_oid(previous_cmoids, cmoid_p)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("\"%s\" compression access method cannot be preserved", cmname_p))); + + /* + * Remove from previous list, also protect from duplicate + * entries in the PRESERVE list + */ + previous_cmoids = list_delete_oid(previous_cmoids, cmoid_p); + } + + /* delete the current cmoid from the list */ + previous_cmoids = list_delete_oid(previous_cmoids, cmoid); + + /* + * If the list of previous Oids is not empty after deletions then + * we need to rewrite tuples in the table. Also remove the dependency + * on the old compression methods which are no longer preserved. + */ + if (list_length(previous_cmoids) != 0) + { + remove_old_dependencies(att->attrelid, att->attnum, + previous_cmoids); + *need_rewrite = true; + } + + /* Cleanup */ + list_free(previous_cmoids); + } + + return cmoid; +} + +/* + * Construct ColumnCompression node from the compression method oid. + */ +ColumnCompression * +MakeColumnCompression(Oid attcompression) +{ + ColumnCompression *node; + + if (!OidIsValid(attcompression)) + return NULL; + + node = makeNode(ColumnCompression); + node->cmname = get_am_name(attcompression); + + return node; +} diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 586a92f0c1..2a1841c353 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -530,7 +530,9 @@ static void ATExecEnableRowSecurity(Relation rel); static void ATExecDisableRowSecurity(Relation rel); static void ATExecForceNoForceRowSecurity(Relation rel, bool force_rls); static ObjectAddress ATExecSetCompression(AlteredTableInfo *tab, Relation rel, - const char *column, Node *newValue, LOCKMODE lockmode); + const char *column, + ColumnCompression *compression, + LOCKMODE lockmode); static void index_copy_data(Relation rel, RelFileNode newrnode); static const char *storage_name(char c); @@ -562,7 +564,6 @@ static void refuseDupeIndexAttach(Relation parentIdx, Relation partIdx, static List *GetParentedForeignKeyRefs(Relation partition); static void ATDetachCheckNoForeignKeyRefs(Relation partition); static void ATExecAlterCollationRefreshVersion(Relation rel, List *coll); -static Oid GetAttributeCompression(Form_pg_attribute att, char *compression); /* ---------------------------------------------------------------- * DefineRelation @@ -587,6 +588,7 @@ ObjectAddress DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, ObjectAddress *typaddress, const char *queryString) { + int i; char relname[NAMEDATALEN]; Oid namespaceId; Oid relationId; @@ -865,7 +867,7 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, relkind == RELKIND_PARTITIONED_TABLE || relkind == RELKIND_MATVIEW) attr->attcompression = - GetAttributeCompression(attr, colDef->compression); + GetAttributeCompression(attr, colDef->compression, NULL); else attr->attcompression = InvalidOid; } @@ -935,6 +937,20 @@ DefineRelation(CreateStmt *stmt, char relkind, Oid ownerId, */ rel = relation_open(relationId, AccessExclusiveLock); + /* + * Add the dependency on the respective compression AM for the relation + * attributes. + */ + for (i = 0; i < (RelationGetDescr(rel))->natts; i++) + { + Form_pg_attribute attr; + + attr = TupleDescAttr(RelationGetDescr(rel), i); + if (OidIsValid(attr->attcompression)) + add_column_compression_dependency(attr->attrelid, attr->attnum, + attr->attcompression); + } + /* * Now add any newly specified column default and generation expressions * to the new relation. These are passed to us in the form of raw @@ -2415,16 +2431,17 @@ MergeAttributes(List *schema, List *supers, char relpersistence, /* Copy/check compression parameter */ if (OidIsValid(attribute->attcompression)) { - char *compression = get_am_name(attribute->attcompression); + ColumnCompression *compression = + MakeColumnCompression(attribute->attcompression); if (!def->compression) def->compression = compression; - else if (strcmp(def->compression, compression) != 0) + else if (strcmp(def->compression->cmname, compression->cmname)) ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("column \"%s\" has a compression method conflict", attributeName), - errdetail("%s versus %s", def->compression, compression))); + errdetail("%s versus %s", def->compression->cmname, compression->cmname))); } def->inhcount++; @@ -2461,7 +2478,8 @@ MergeAttributes(List *schema, List *supers, char relpersistence, def->collOid = attribute->attcollation; def->constraints = NIL; def->location = -1; - def->compression = get_am_name(attribute->attcompression); + def->compression = MakeColumnCompression( + attribute->attcompression); inhSchema = lappend(inhSchema, def); newattmap->attnums[parent_attno - 1] = ++child_attno; } @@ -2712,12 +2730,12 @@ MergeAttributes(List *schema, List *supers, char relpersistence, def->compression = newdef->compression; else if (newdef->compression) { - if (strcmp(def->compression, newdef->compression)) + if (strcmp(def->compression->cmname, newdef->compression->cmname)) ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("column \"%s\" has a compression method conflict", attributeName), - errdetail("%s versus %s", def->compression, newdef->compression))); + errdetail("%s versus %s", def->compression->cmname, newdef->compression->cmname))); } /* Mark the column as locally defined */ @@ -4908,7 +4926,8 @@ ATExecCmd(List **wqueue, AlteredTableInfo *tab, Relation rel, ATExecAlterCollationRefreshVersion(rel, cmd->object); break; case AT_SetCompression: - address = ATExecSetCompression(tab, rel, cmd->name, cmd->def, + address = ATExecSetCompression(tab, rel, cmd->name, + (ColumnCompression *) cmd->def, lockmode); break; default: /* oops */ @@ -6414,7 +6433,8 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel, if (rel->rd_rel->relkind == RELKIND_RELATION || rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) attribute.attcompression = GetAttributeCompression(&attribute, - colDef->compression); + colDef->compression, + NULL); else attribute.attcompression = InvalidOid; @@ -6589,6 +6609,7 @@ ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel, */ add_column_datatype_dependency(myrelid, newattnum, attribute.atttypid); add_column_collation_dependency(myrelid, newattnum, attribute.attcollation); + add_column_compression_dependency(myrelid, newattnum, attribute.attcompression); /* * Propagate to children as appropriate. Unlike most other ALTER @@ -6736,6 +6757,28 @@ add_column_collation_dependency(Oid relid, int32 attnum, Oid collid) } } +/* + * Install a dependency for compression on its column. + * + * This is used for identifying all the supported compression methods + * (current and preserved) for a attribute. + * + * If dependency is already there the whole thing is skipped. + */ +void +add_column_compression_dependency(Oid relid, int32 attnum, Oid cmoid) +{ + ObjectAddress acref, + attref; + + Assert(relid > 0 && attnum > 0); + + ObjectAddressSet(acref, AccessMethodRelationId, cmoid); + ObjectAddressSubSet(attref, RelationRelationId, relid, attnum); + + recordMultipleDependencies(&attref, &acref, 1, DEPENDENCY_NORMAL, true); +} + /* * ALTER TABLE ALTER COLUMN DROP NOT NULL */ @@ -11867,7 +11910,8 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel, foundDep->refobjid == attTup->attcollation) && !(foundDep->refclassid == RelationRelationId && foundDep->refobjid == RelationGetRelid(rel) && - foundDep->refobjsubid != 0) + foundDep->refobjsubid != 0) && + foundDep->refclassid != AccessMethodRelationId ) elog(ERROR, "found unexpected dependency for column: %s", getObjectDescription(&foundObject, false)); @@ -11982,6 +12026,11 @@ ATExecAlterColumnType(AlteredTableInfo *tab, Relation rel, add_column_datatype_dependency(RelationGetRelid(rel), attnum, targettype); add_column_collation_dependency(RelationGetRelid(rel), attnum, targetcollid); + /* Create dependency for new attribute compression */ + if (OidIsValid(attTup->attcompression)) + add_column_compression_dependency(RelationGetRelid(rel), attnum, + attTup->attcompression); + /* * Drop any pg_statistic entry for the column, since it's now wrong type */ @@ -15086,24 +15135,21 @@ static ObjectAddress ATExecSetCompression(AlteredTableInfo *tab, Relation rel, const char *column, - Node *newValue, + ColumnCompression *compression, LOCKMODE lockmode) { Relation attrel; HeapTuple tuple; Form_pg_attribute atttableform; AttrNumber attnum; - char *compression; char typstorage; Oid cmoid; + bool need_rewrite; Datum values[Natts_pg_attribute]; bool nulls[Natts_pg_attribute]; bool replace[Natts_pg_attribute]; ObjectAddress address; - Assert(IsA(newValue, String)); - compression = strVal(newValue); - attrel = table_open(AttributeRelationId, RowExclusiveLock); tuple = SearchSysCacheAttName(RelationGetRelid(rel), column); @@ -15136,11 +15182,16 @@ ATExecSetCompression(AlteredTableInfo *tab, memset(replace, false, sizeof(replace)); /* Get the attribute compression method. */ - cmoid = GetAttributeCompression(atttableform, compression); + cmoid = GetAttributeCompression(atttableform, compression, &need_rewrite); if (atttableform->attcompression != cmoid) + add_column_compression_dependency(atttableform->attrelid, + atttableform->attnum, cmoid); + if (need_rewrite) tab->rewrite |= AT_REWRITE_ALTER_COMPRESSION; + atttableform->attcompression = cmoid; + atttableform->attcompression = cmoid; CatalogTupleUpdate(attrel, &tuple->t_self, tuple); @@ -17865,42 +17916,3 @@ ATExecAlterCollationRefreshVersion(Relation rel, List *coll) index_update_collation_versions(rel->rd_id, get_collation_oid(coll, false)); CacheInvalidateRelcache(rel); } - -/* - * resolve column compression specification to an OID. - */ -static Oid -GetAttributeCompression(Form_pg_attribute att, char *compression) -{ - char typstorage = get_typstorage(att->atttypid); - Oid amoid; - - /* - * No compression for the plain/external storage, refer comments atop - * attcompression parameter in pg_attribute.h - */ - if (!IsStorageCompressible(typstorage)) - { - if (compression == NULL) - return InvalidOid; - - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("column data type %s does not support compression", - format_type_be(att->atttypid)))); - } - - /* fallback to default compression if it's not specified */ - if (compression == NULL) - return GetDefaultToastCompression(); - - amoid = get_compression_am_oid(compression, false); - -#ifndef HAVE_LIBLZ4 - if (amoid == LZ4_COMPRESSION_AM_OID) - ereport(ERROR, - (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), - errmsg("not built with lz4 support"))); -#endif - return amoid; -} diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index ea82a05591..90d092671e 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -44,6 +44,7 @@ #include "access/tableam.h" #include "access/xact.h" #include "catalog/catalog.h" +#include "commands/defrem.h" #include "commands/trigger.h" #include "executor/execPartition.h" #include "executor/executor.h" @@ -2068,8 +2069,8 @@ CompareCompressionMethodAndDecompress(TupleTableSlot *slot, /* * Loop over all the attributes in the tuple and check if any attribute is - * compressed and its compression method is not same as the target - * atrribute's compression method then decompress it. + * compressed and its compression method is not is not supported by the + * target attribute then we need to decompress */ for (i = 0; i < natts; i++) { @@ -2094,12 +2095,13 @@ CompareCompressionMethodAndDecompress(TupleTableSlot *slot, DatumGetPointer(slot->tts_values[attnum - 1]); /* - * Get the compression method Oid stored in the toast header and - * compare it with the compression method of the target. + * Get the compression method stored in the toast header and if the + * compression method is not supported by the target attribute then + * we need to decompress it. */ cmoid = toast_get_compression_oid(new_value); if (OidIsValid(cmoid) && - targetTupDesc->attrs[i].attcompression != cmoid) + !IsCompressionSupported(&targetTupDesc->attrs[i], cmoid)) { new_value = detoast_attr(new_value); slot->tts_values[attnum - 1] = PointerGetDatum(new_value); diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 1338e04409..6a11f8eb60 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -2966,7 +2966,7 @@ _copyColumnDef(const ColumnDef *from) COPY_STRING_FIELD(colname); COPY_NODE_FIELD(typeName); - COPY_STRING_FIELD(compression); + COPY_NODE_FIELD(compression); COPY_SCALAR_FIELD(inhcount); COPY_SCALAR_FIELD(is_local); COPY_SCALAR_FIELD(is_not_null); @@ -2986,6 +2986,18 @@ _copyColumnDef(const ColumnDef *from) return newnode; } +static ColumnCompression * +_copyColumnCompression(const ColumnCompression *from) +{ + ColumnCompression *newnode = makeNode(ColumnCompression); + + COPY_STRING_FIELD(cmname); + COPY_SCALAR_FIELD(preserve_all); + COPY_NODE_FIELD(preserve); + + return newnode; +} + static Constraint * _copyConstraint(const Constraint *from) { @@ -5675,6 +5687,9 @@ copyObjectImpl(const void *from) case T_ColumnDef: retval = _copyColumnDef(from); break; + case T_ColumnCompression: + retval = _copyColumnCompression(from); + break; case T_Constraint: retval = _copyConstraint(from); break; diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c index f3592003da..26a9b85974 100644 --- a/src/backend/nodes/equalfuncs.c +++ b/src/backend/nodes/equalfuncs.c @@ -2599,7 +2599,7 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b) { COMPARE_STRING_FIELD(colname); COMPARE_NODE_FIELD(typeName); - COMPARE_STRING_FIELD(compression); + COMPARE_NODE_FIELD(compression); COMPARE_SCALAR_FIELD(inhcount); COMPARE_SCALAR_FIELD(is_local); COMPARE_SCALAR_FIELD(is_not_null); @@ -2619,6 +2619,16 @@ _equalColumnDef(const ColumnDef *a, const ColumnDef *b) return true; } +static bool +_equalColumnCompression(const ColumnCompression *a, const ColumnCompression *b) +{ + COMPARE_STRING_FIELD(cmname); + COMPARE_SCALAR_FIELD(preserve_all); + COMPARE_NODE_FIELD(preserve); + + return true; +} + static bool _equalConstraint(const Constraint *a, const Constraint *b) { @@ -3724,6 +3734,9 @@ equal(const void *a, const void *b) case T_ColumnDef: retval = _equalColumnDef(a, b); break; + case T_ColumnCompression: + retval = _equalColumnCompression(a, b); + break; case T_Constraint: retval = _equalConstraint(a, b); break; diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c index 0605ef3f84..b584a58ba3 100644 --- a/src/backend/nodes/outfuncs.c +++ b/src/backend/nodes/outfuncs.c @@ -2863,7 +2863,7 @@ _outColumnDef(StringInfo str, const ColumnDef *node) WRITE_STRING_FIELD(colname); WRITE_NODE_FIELD(typeName); - WRITE_STRING_FIELD(compression); + WRITE_NODE_FIELD(compression); WRITE_INT_FIELD(inhcount); WRITE_BOOL_FIELD(is_local); WRITE_BOOL_FIELD(is_not_null); @@ -2881,6 +2881,16 @@ _outColumnDef(StringInfo str, const ColumnDef *node) WRITE_LOCATION_FIELD(location); } +static void +_outColumnCompression(StringInfo str, const ColumnCompression *node) +{ + WRITE_NODE_TYPE("COLUMNCOMPRESSION"); + + WRITE_STRING_FIELD(cmname); + WRITE_BOOL_FIELD(preserve_all); + WRITE_NODE_FIELD(preserve); +} + static void _outTypeName(StringInfo str, const TypeName *node) { @@ -4258,6 +4268,9 @@ outNode(StringInfo str, const void *obj) case T_ColumnDef: _outColumnDef(str, obj); break; + case T_ColumnCompression: + _outColumnCompression(str, obj); + break; case T_TypeName: _outTypeName(str, obj); break; diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 30acfe615d..9eb2b04d58 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -596,7 +596,9 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type <list> hash_partbound %type <defelt> hash_partbound_elem -%type <str> optColumnCompression +%type <node> optColumnCompression alterColumnCompression +%type <str> compressionClause +%type <list> optCompressionPreserve /* * Non-keyword token types. These are hard-wired into the "flex" lexer. @@ -2309,12 +2311,12 @@ alter_table_cmd: $$ = (Node *)n; } /* ALTER TABLE <name> ALTER [COLUMN] <colname> SET (COMPRESSION <cm>) */ - | ALTER opt_column ColId SET optColumnCompression + | ALTER opt_column ColId SET alterColumnCompression { AlterTableCmd *n = makeNode(AlterTableCmd); n->subtype = AT_SetCompression; n->name = $3; - n->def = (Node *) makeString($5); + n->def = $5; $$ = (Node *)n; } /* ALTER TABLE <name> DROP [COLUMN] IF EXISTS <colname> [RESTRICT|CASCADE] */ @@ -3437,7 +3439,7 @@ columnDef: ColId Typename optColumnCompression create_generic_options ColQualLis ColumnDef *n = makeNode(ColumnDef); n->colname = $1; n->typeName = $2; - n->compression = $3; + n->compression = (ColumnCompression *) $3; n->inhcount = 0; n->is_local = true; n->is_not_null = false; @@ -3492,13 +3494,43 @@ columnOptions: ColId ColQualList } ; +optCompressionPreserve: + PRESERVE '(' name_list ')' { $$ = $3; } + | /*EMPTY*/ { $$ = NULL; } + ; + +compressionClause: + COMPRESSION name { $$ = pstrdup($2); } + ; + optColumnCompression: - COMPRESSION name - { - $$ = $2; - } - | /*EMPTY*/ { $$ = NULL; } - ; + compressionClause + { + ColumnCompression *n = makeNode(ColumnCompression); + n->cmname = $1; + n->preserve = NIL; + $$ = (Node *) n; + } + | /*EMPTY*/ { $$ = NULL; } + ; + +alterColumnCompression: + compressionClause optCompressionPreserve + { + ColumnCompression *n = makeNode(ColumnCompression); + n->cmname = $1; + n->preserve = (List *) $2; + $$ = (Node *) n; + } + | compressionClause PRESERVE ALL + { + ColumnCompression *n = makeNode(ColumnCompression); + n->cmname = $1; + n->preserve_all = true; + n->preserve = NIL; + $$ = (Node *) n; + } + ; ColQualList: ColQualList ColConstraint { $$ = lappend($1, $2); } diff --git a/src/backend/parser/parse_utilcmd.c b/src/backend/parser/parse_utilcmd.c index cf4413da64..45f4724a13 100644 --- a/src/backend/parser/parse_utilcmd.c +++ b/src/backend/parser/parse_utilcmd.c @@ -1086,7 +1086,7 @@ transformTableLikeClause(CreateStmtContext *cxt, TableLikeClause *table_like_cla /* Likewise, copy compression if requested */ if ((table_like_clause->options & CREATE_TABLE_LIKE_COMPRESSION) != 0 && OidIsValid(attribute->attcompression)) - def->compression = get_am_name(attribute->attcompression); + def->compression = MakeColumnCompression(attribute->attcompression); else def->compression = NULL; diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 46044cb92a..7bf345a4ac 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -9037,6 +9037,80 @@ getTableAttrs(Archive *fout, TableInfo *tblinfo, int numTables) } PQclear(res); } + + /* + * Get compression info + */ + if (fout->remoteVersion >= 140000 && dopt->binary_upgrade) + { + int i_amname; + int i_amoid; + int i_curattnum; + int start; + + pg_log_info("finding compression info for table \"%s.%s\"", + tbinfo->dobj.namespace->dobj.name, + tbinfo->dobj.name); + + tbinfo->attcompression = pg_malloc0(tbinfo->numatts * sizeof(AttrCompressionInfo *)); + + resetPQExpBuffer(q); + appendPQExpBuffer(q, + " SELECT attrelid::pg_catalog.regclass AS relname, attname," + " amname, am.oid as amoid, d.objsubid AS curattnum" + " FROM pg_depend d" + " JOIN pg_attribute a ON" + " (classid = 'pg_class'::pg_catalog.regclass::pg_catalog.oid AND a.attrelid = d.objid" + " AND a.attnum = d.objsubid AND d.deptype = 'n'" + " AND d.refclassid = 'pg_am'::pg_catalog.regclass::pg_catalog.oid)" + " JOIN pg_am am ON" + " (d.deptype = 'n' AND d.refobjid = am.oid)" + " WHERE (deptype = 'n' AND d.objid = %d AND a.attcompression != am.oid);", + tbinfo->dobj.catId.oid); + + res = ExecuteSqlQuery(fout, q->data, PGRES_TUPLES_OK); + ntups = PQntuples(res); + + if (ntups > 0) + { + int j; + int k; + + i_amname = PQfnumber(res, "amname"); + i_amoid = PQfnumber(res, "amoid"); + i_curattnum = PQfnumber(res, "curattnum"); + + start = 0; + + for (j = 0; j < ntups; j++) + { + int attnum = atoi(PQgetvalue(res, j, i_curattnum)); + + if ((j == ntups - 1) || atoi(PQgetvalue(res, j + 1, i_curattnum)) != attnum) + { + AttrCompressionInfo *cminfo = pg_malloc(sizeof(AttrCompressionInfo)); + + cminfo->nitems = j - start + 1; + cminfo->items = pg_malloc(sizeof(AttrCompressionItem *) * cminfo->nitems); + + for (k = start; k < start + cminfo->nitems; k++) + { + AttrCompressionItem *cmitem = pg_malloc0(sizeof(AttrCompressionItem)); + + cmitem->amname = pg_strdup(PQgetvalue(res, k, i_amname)); + cmitem->amoid = atooid(PQgetvalue(res, k, i_amoid)); + + cminfo->items[k - start] = cmitem; + } + + tbinfo->attcompression[attnum - 1] = cminfo; + start = j + 1; /* start from next */ + } + } + } + + PQclear(res); + } } destroyPQExpBuffer(q); @@ -16343,6 +16417,33 @@ dumpTableSchema(Archive *fout, const TableInfo *tbinfo) qualrelname, fmtId(tbinfo->attnames[j]), tbinfo->attfdwoptions[j]); + + /* + * Dump per-column compression options + */ + if (tbinfo->attcompression && tbinfo->attcompression[j]) + { + AttrCompressionInfo *cminfo = tbinfo->attcompression[j]; + + appendPQExpBuffer(q, "ALTER TABLE %s ALTER COLUMN %s\nSET COMPRESSION %s", + qualrelname, fmtId(tbinfo->attnames[j]), tbinfo->attcmnames[j]); + + if (cminfo->nitems > 0) + { + appendPQExpBuffer(q, "\nPRESERVE ("); + for (int i = 0; i < cminfo->nitems; i++) + { + AttrCompressionItem *item = cminfo->items[i]; + + if (i == 0) + appendPQExpBuffer(q, "%s", item->amname); + else + appendPQExpBuffer(q, ", %s", item->amname); + } + appendPQExpBuffer(q, ")"); + } + appendPQExpBuffer(q, ";\n"); + } } if (ftoptions) diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h index 1789e18f46..a829528cd0 100644 --- a/src/bin/pg_dump/pg_dump.h +++ b/src/bin/pg_dump/pg_dump.h @@ -327,7 +327,8 @@ typedef struct _tableInfo bool needs_override; /* has GENERATED ALWAYS AS IDENTITY */ char *amname; /* relation access method */ char **attcmnames; /* per-attribute current compression method */ - + struct _attrCompressionInfo **attcompression; /* per-attribute all + compression data */ /* * Stuff computed only for dumpable tables. */ @@ -356,6 +357,18 @@ typedef struct _attrDefInfo bool separate; /* true if must dump as separate item */ } AttrDefInfo; +typedef struct _attrCompressionItem +{ + Oid amoid; /* attribute compression oid */ + char *amname; /* compression access method name */ +} AttrCompressionItem; + +typedef struct _attrCompressionInfo +{ + int nitems; + AttrCompressionItem **items; +} AttrCompressionInfo; + typedef struct _tableDataInfo { DumpableObject dobj; diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c index ffa8d05edf..869fd3676d 100644 --- a/src/bin/psql/tab-complete.c +++ b/src/bin/psql/tab-complete.c @@ -2103,6 +2103,13 @@ psql_completion(const char *text, int start, int end) else if (Matches("ALTER", "TABLE", MatchAny, "ALTER", "COLUMN", MatchAny, "SET", "(") || Matches("ALTER", "TABLE", MatchAny, "ALTER", MatchAny, "SET", "(")) COMPLETE_WITH("n_distinct", "n_distinct_inherited"); + /* ALTER TABLE ALTER [COLUMN] <foo> SET COMPRESSION */ + else if (Matches("ALTER", "TABLE", MatchAny, "ALTER", "COLUMN", MatchAny, "SET", "COMPRESSION", MatchAny) || + Matches("ALTER", "TABLE", MatchAny, "ALTER", MatchAny, "SET", "COMPRESSION", MatchAny)) + COMPLETE_WITH("PRESERVE"); + else if (Matches("ALTER", "TABLE", MatchAny, "ALTER", "COLUMN", MatchAny, "SET", "COMPRESSION", MatchAny, "PRESERVE") || + Matches("ALTER", "TABLE", MatchAny, "ALTER", MatchAny, "SET", "COMPRESSION", MatchAny, "PRESERVE")) + COMPLETE_WITH("( ", "ALL"); /* ALTER TABLE ALTER [COLUMN] <foo> SET STORAGE */ else if (Matches("ALTER", "TABLE", MatchAny, "ALTER", "COLUMN", MatchAny, "SET", "STORAGE") || Matches("ALTER", "TABLE", MatchAny, "ALTER", MatchAny, "SET", "STORAGE")) diff --git a/src/include/commands/defrem.h b/src/include/commands/defrem.h index e5aea8a240..bd53f9bb0f 100644 --- a/src/include/commands/defrem.h +++ b/src/include/commands/defrem.h @@ -143,6 +143,13 @@ extern Oid get_compression_am_oid(const char *amname, bool missing_ok); extern Oid get_am_oid(const char *amname, bool missing_ok); extern char *get_am_name(Oid amOid); +/* commands/compressioncmds.c */ +extern Oid GetAttributeCompression(Form_pg_attribute att, + ColumnCompression *compression, + bool *need_rewrite); +extern ColumnCompression *MakeColumnCompression(Oid atttcompression); +extern bool IsCompressionSupported(Form_pg_attribute att, Oid cmoid); + /* support routines in commands/define.c */ extern char *defGetString(DefElem *def); diff --git a/src/include/commands/tablecmds.h b/src/include/commands/tablecmds.h index b3d30acc35..e6c98e65d4 100644 --- a/src/include/commands/tablecmds.h +++ b/src/include/commands/tablecmds.h @@ -97,5 +97,7 @@ extern void RangeVarCallbackOwnsRelation(const RangeVar *relation, Oid relId, Oid oldRelId, void *arg); extern bool PartConstraintImpliedByRelConstraint(Relation scanrel, List *partConstraint); +extern void add_column_compression_dependency(Oid relid, int32 attnum, + Oid cmoid); #endif /* TABLECMDS_H */ diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h index 20d6f96f62..24deaad253 100644 --- a/src/include/nodes/nodes.h +++ b/src/include/nodes/nodes.h @@ -481,6 +481,7 @@ typedef enum NodeTag T_PartitionBoundSpec, T_PartitionRangeDatum, T_PartitionCmd, + T_ColumnCompression, T_VacuumRelation, /* diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index f9a87dee02..ce0913e18a 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -623,6 +623,20 @@ typedef struct RangeTableSample int location; /* method name location, or -1 if unknown */ } RangeTableSample; +/* + * ColumnCompression - compression parameters for some attribute + * + * This represents compression information defined using clause: + * .. COMPRESSION <compression method> PRESERVE <compression methods> + */ +typedef struct ColumnCompression +{ + NodeTag type; + char *cmname; + bool preserve_all; + List *preserve; +} ColumnCompression; + /* * ColumnDef - column definition (used in various creates) * @@ -646,7 +660,7 @@ typedef struct ColumnDef NodeTag type; char *colname; /* name of column */ TypeName *typeName; /* type of column */ - char *compression; /* compression method for column */ + ColumnCompression *compression; /* column compression */ int inhcount; /* number of times column is inherited */ bool is_local; /* column has local (non-inherited) def'n */ bool is_not_null; /* NOT NULL constraint specified? */ diff --git a/src/test/regress/expected/compression.out b/src/test/regress/expected/compression.out index 21c1b451d2..3ed33b6534 100644 --- a/src/test/regress/expected/compression.out +++ b/src/test/regress/expected/compression.out @@ -226,12 +226,47 @@ SELECT pg_column_compression(f1) FROM cmpart; lz4 (2 rows) +-- preserve old compression method +ALTER TABLE cmdata ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (lz4); +INSERT INTO cmdata VALUES (repeat('1234567890',1004)); +\d+ cmdata + Table "public.cmdata" + Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description +--------+------+-----------+----------+---------+----------+-------------+--------------+------------- + f1 | text | | | | extended | pglz | | +Indexes: + "idx" btree (f1) + +SELECT pg_column_compression(f1) FROM cmdata; + pg_column_compression +----------------------- + lz4 + pglz +(2 rows) + +\d+ cmdata + Table "public.cmdata" + Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description +--------+------+-----------+----------+---------+----------+-------------+--------------+------------- + f1 | text | | | | extended | pglz | | +Indexes: + "idx" btree (f1) + +ALTER TABLE cmdata ALTER COLUMN f1 SET COMPRESSION lz4 PRESERVE ALL; +SELECT pg_column_compression(f1) FROM cmdata; + pg_column_compression +----------------------- + lz4 + pglz +(2 rows) + -- check data is ok SELECT length(f1) FROM cmdata; length -------- 10000 -(1 row) + 10040 +(2 rows) SELECT length(f1) FROM cmdata1; length diff --git a/src/test/regress/expected/compression_1.out b/src/test/regress/expected/compression_1.out index 64c5855bf7..36a5f8ba5e 100644 --- a/src/test/regress/expected/compression_1.out +++ b/src/test/regress/expected/compression_1.out @@ -205,12 +205,49 @@ SELECT pg_column_compression(f1) FROM cmpart; ERROR: relation "cmpart" does not exist LINE 1: SELECT pg_column_compression(f1) FROM cmpart; ^ +-- preserve old compression method +ALTER TABLE cmdata ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (lz4); +ERROR: "lz4" compression access method cannot be preserved +INSERT INTO cmdata VALUES (repeat('1234567890',1004)); +\d+ cmdata + Table "public.cmdata" + Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description +--------+------+-----------+----------+---------+----------+-------------+--------------+------------- + f1 | text | | | | extended | pglz | | +Indexes: + "idx" btree (f1) + +SELECT pg_column_compression(f1) FROM cmdata; + pg_column_compression +----------------------- + pglz + pglz +(2 rows) + +\d+ cmdata + Table "public.cmdata" + Column | Type | Collation | Nullable | Default | Storage | Compression | Stats target | Description +--------+------+-----------+----------+---------+----------+-------------+--------------+------------- + f1 | text | | | | extended | pglz | | +Indexes: + "idx" btree (f1) + +ALTER TABLE cmdata ALTER COLUMN f1 SET COMPRESSION lz4 PRESERVE ALL; +ERROR: not built with lz4 support +SELECT pg_column_compression(f1) FROM cmdata; + pg_column_compression +----------------------- + pglz + pglz +(2 rows) + -- check data is ok SELECT length(f1) FROM cmdata; length -------- 10000 -(1 row) + 10040 +(2 rows) SELECT length(f1) FROM cmdata1; ERROR: relation "cmdata1" does not exist diff --git a/src/test/regress/expected/create_index.out b/src/test/regress/expected/create_index.out index 830fdddf24..8f984510ac 100644 --- a/src/test/regress/expected/create_index.out +++ b/src/test/regress/expected/create_index.out @@ -2066,19 +2066,21 @@ WHERE classid = 'pg_class'::regclass AND 'concur_reindex_ind4'::regclass, 'concur_reindex_matview'::regclass) ORDER BY 1, 2; - obj | objref | deptype -------------------------------------------+------------------------------------------------------------+--------- - index concur_reindex_ind1 | constraint concur_reindex_ind1 on table concur_reindex_tab | i - index concur_reindex_ind2 | collation "default" | n - index concur_reindex_ind2 | column c2 of table concur_reindex_tab | a - index concur_reindex_ind3 | column c1 of table concur_reindex_tab | a - index concur_reindex_ind3 | table concur_reindex_tab | a - index concur_reindex_ind4 | collation "default" | n - index concur_reindex_ind4 | column c1 of table concur_reindex_tab | a - index concur_reindex_ind4 | column c2 of table concur_reindex_tab | a - materialized view concur_reindex_matview | schema public | n - table concur_reindex_tab | schema public | n -(10 rows) + obj | objref | deptype +-------------------------------------------------------+------------------------------------------------------------+--------- + column c2 of materialized view concur_reindex_matview | access method pglz | n + column c2 of table concur_reindex_tab | access method pglz | n + index concur_reindex_ind1 | constraint concur_reindex_ind1 on table concur_reindex_tab | i + index concur_reindex_ind2 | collation "default" | n + index concur_reindex_ind2 | column c2 of table concur_reindex_tab | a + index concur_reindex_ind3 | column c1 of table concur_reindex_tab | a + index concur_reindex_ind3 | table concur_reindex_tab | a + index concur_reindex_ind4 | collation "default" | n + index concur_reindex_ind4 | column c1 of table concur_reindex_tab | a + index concur_reindex_ind4 | column c2 of table concur_reindex_tab | a + materialized view concur_reindex_matview | schema public | n + table concur_reindex_tab | schema public | n +(12 rows) REINDEX INDEX CONCURRENTLY concur_reindex_ind1; REINDEX TABLE CONCURRENTLY concur_reindex_tab; @@ -2095,19 +2097,21 @@ WHERE classid = 'pg_class'::regclass AND 'concur_reindex_ind4'::regclass, 'concur_reindex_matview'::regclass) ORDER BY 1, 2; - obj | objref | deptype -------------------------------------------+------------------------------------------------------------+--------- - index concur_reindex_ind1 | constraint concur_reindex_ind1 on table concur_reindex_tab | i - index concur_reindex_ind2 | collation "default" | n - index concur_reindex_ind2 | column c2 of table concur_reindex_tab | a - index concur_reindex_ind3 | column c1 of table concur_reindex_tab | a - index concur_reindex_ind3 | table concur_reindex_tab | a - index concur_reindex_ind4 | collation "default" | n - index concur_reindex_ind4 | column c1 of table concur_reindex_tab | a - index concur_reindex_ind4 | column c2 of table concur_reindex_tab | a - materialized view concur_reindex_matview | schema public | n - table concur_reindex_tab | schema public | n -(10 rows) + obj | objref | deptype +-------------------------------------------------------+------------------------------------------------------------+--------- + column c2 of materialized view concur_reindex_matview | access method pglz | n + column c2 of table concur_reindex_tab | access method pglz | n + index concur_reindex_ind1 | constraint concur_reindex_ind1 on table concur_reindex_tab | i + index concur_reindex_ind2 | collation "default" | n + index concur_reindex_ind2 | column c2 of table concur_reindex_tab | a + index concur_reindex_ind3 | column c1 of table concur_reindex_tab | a + index concur_reindex_ind3 | table concur_reindex_tab | a + index concur_reindex_ind4 | collation "default" | n + index concur_reindex_ind4 | column c1 of table concur_reindex_tab | a + index concur_reindex_ind4 | column c2 of table concur_reindex_tab | a + materialized view concur_reindex_matview | schema public | n + table concur_reindex_tab | schema public | n +(12 rows) -- Check that comments are preserved CREATE TABLE testcomment (i int); diff --git a/src/test/regress/sql/compression.sql b/src/test/regress/sql/compression.sql index b9daa33b74..5774b55d82 100644 --- a/src/test/regress/sql/compression.sql +++ b/src/test/regress/sql/compression.sql @@ -96,6 +96,15 @@ ALTER TABLE cmpart1 ALTER COLUMN f1 SET COMPRESSION pglz; ALTER TABLE cmpart2 ALTER COLUMN f1 SET COMPRESSION lz4; SELECT pg_column_compression(f1) FROM cmpart; +-- preserve old compression method +ALTER TABLE cmdata ALTER COLUMN f1 SET COMPRESSION pglz PRESERVE (lz4); +INSERT INTO cmdata VALUES (repeat('1234567890',1004)); +\d+ cmdata +SELECT pg_column_compression(f1) FROM cmdata; +\d+ cmdata +ALTER TABLE cmdata ALTER COLUMN f1 SET COMPRESSION lz4 PRESERVE ALL; +SELECT pg_column_compression(f1) FROM cmdata; + -- check data is ok SELECT length(f1) FROM cmdata; SELECT length(f1) FROM cmdata1; -- 2.17.0 --YZ5djTAD1cGYuMQK-- ^ permalink raw reply [nested|flat] 2+ messages in thread
* Re: MERGE ... WHEN NOT MATCHED BY SOURCE @ 2023-01-02 18:24 Dean Rasheed <[email protected]> 0 siblings, 0 replies; 2+ messages in thread From: Dean Rasheed @ 2023-01-02 18:24 UTC (permalink / raw) To: PostgreSQL Hackers <[email protected]> On Fri, 30 Dec 2022 at 16:56, Dean Rasheed <[email protected]> wrote: > > Attached is a WIP patch. > Updated patch attached, now with updated docs and some other minor tidying up. Regards, Dean Attachments: [text/x-patch] support-merge-when-not-matched-by-source-v2.patch (34.1K, ../../CAEZATCUm0yku11PQ61d2M92sWnLJaRqvrMfrcmsLCeap1sSdEA@mail.gmail.com/2-support-merge-when-not-matched-by-source-v2.patch) download | inline diff: diff --git a/doc/src/sgml/mvcc.sgml b/doc/src/sgml/mvcc.sgml new file mode 100644 index b87ad5c..1482ede --- a/doc/src/sgml/mvcc.sgml +++ b/doc/src/sgml/mvcc.sgml @@ -396,8 +396,8 @@ originally matched appears later in the list of actions. On the other hand, if the row is concurrently updated or deleted so that the join condition fails, then <command>MERGE</command> will - evaluate the condition's <literal>NOT MATCHED</literal> actions next, - and execute the first one that succeeds. + evaluate the condition's <literal>NOT MATCHED [BY TARGET]</literal> + actions next, and execute the first one that succeeds. If <command>MERGE</command> attempts an <command>INSERT</command> and a unique index is present and a duplicate row is concurrently inserted, then a uniqueness violation error is raised; diff --git a/doc/src/sgml/ref/merge.sgml b/doc/src/sgml/ref/merge.sgml new file mode 100644 index 0995fe0..8ef121a --- a/doc/src/sgml/ref/merge.sgml +++ b/doc/src/sgml/ref/merge.sgml @@ -33,7 +33,8 @@ USING <replaceable class="parameter">dat <phrase>and <replaceable class="parameter">when_clause</replaceable> is:</phrase> { WHEN MATCHED [ AND <replaceable class="parameter">condition</replaceable> ] THEN { <replaceable class="parameter">merge_update</replaceable> | <replaceable class="parameter">merge_delete</replaceable> | DO NOTHING } | - WHEN NOT MATCHED [ AND <replaceable class="parameter">condition</replaceable> ] THEN { <replaceable class="parameter">merge_insert</replaceable> | DO NOTHING } } + WHEN NOT MATCHED BY SOURCE [ AND <replaceable class="parameter">condition</replaceable> ] THEN { <replaceable class="parameter">merge_update</replaceable> | <replaceable class="parameter">merge_delete</replaceable> | DO NOTHING } | + WHEN NOT MATCHED [BY TARGET] [ AND <replaceable class="parameter">condition</replaceable> ] THEN { <replaceable class="parameter">merge_insert</replaceable> | DO NOTHING } } <phrase>and <replaceable class="parameter">merge_insert</replaceable> is:</phrase> @@ -70,7 +71,9 @@ DELETE from <replaceable class="parameter">data_source</replaceable> to <replaceable class="parameter">target_table_name</replaceable> producing zero or more candidate change rows. For each candidate change - row, the status of <literal>MATCHED</literal> or <literal>NOT MATCHED</literal> + row, the status of <literal>MATCHED</literal>, + <literal>NOT MATCHED BY SOURCE</literal>, + or <literal>NOT MATCHED [BY TARGET]</literal> is set just once, after which <literal>WHEN</literal> clauses are evaluated in the order specified. For each candidate change row, the first clause to evaluate as true is executed. No more than one <literal>WHEN</literal> @@ -226,16 +229,37 @@ DELETE At least one <literal>WHEN</literal> clause is required. </para> <para> + The <literal>WHEN</literal> clause may specify <literal>WHEN MATCHED</literal>, + <literal>WHEN NOT MATCHED BY SOURCE</literal>, or + <literal>WHEN NOT MATCHED [BY TARGET]</literal>. + Note that the <acronym>SQL</acronym> standard only defines + <literal>WHEN MATCHED</literal> and <literal>WHEN NOT MATCHED</literal> + (which is defined to mean no matching target row). + <literal>WHEN NOT MATCHED BY SOURCE</literal> is an extension to the + <acronym>SQL</acronym> standard, as is the option to append + <literal>BY TARGET</literal> to <literal>WHEN NOT MATCHED</literal>, to + make its meaning more explicit. + </para> + <para> If the <literal>WHEN</literal> clause specifies <literal>WHEN MATCHED</literal> - and the candidate change row matches a row in the + and the candidate change row matches a row in the source to a row in the <replaceable class="parameter">target_table_name</replaceable>, the <literal>WHEN</literal> clause is executed if the <replaceable class="parameter">condition</replaceable> is absent or it evaluates to <literal>true</literal>. </para> <para> - Conversely, if the <literal>WHEN</literal> clause specifies - <literal>WHEN NOT MATCHED</literal> + If the <literal>WHEN</literal> clause specifies + <literal>WHEN NOT MATCHED BY SOURCE</literal> and the candidate change + row represents a row in the + <replaceable class="parameter">target_table_name</replaceable> that does + not match a source row, the <literal>WHEN</literal> clause is executed + if the <replaceable class="parameter">condition</replaceable> is + absent or it evaluates to <literal>true</literal>. + </para> + <para> + If the <literal>WHEN</literal> clause specifies + <literal>WHEN NOT MATCHED [BY TARGET]</literal> and the candidate change row does not match a row in the <replaceable class="parameter">target_table_name</replaceable>, the <literal>WHEN</literal> clause is executed if the @@ -257,7 +281,10 @@ DELETE <para> A condition on a <literal>WHEN MATCHED</literal> clause can refer to columns in both the source and the target relations. A condition on a - <literal>WHEN NOT MATCHED</literal> clause can only refer to columns from + <literal>WHEN NOT MATCHED BY SOURCE</literal> clause can only refer to + columns from the target relation, since by definition there is no matching + source row. A condition on a <literal>WHEN NOT MATCHED [BY TARGET]</literal> + clause can only refer to columns from the source relation, since by definition there is no matching target row. Only the system attributes from the target table are accessible. </para> @@ -382,8 +409,10 @@ DELETE <literal>WHEN MATCHED</literal> clause, the expression can use values from the original row in the target table, and values from the <literal>data_source</literal> row. - If used in a <literal>WHEN NOT MATCHED</literal> clause, the - expression can use values from the <literal>data_source</literal>. + If used in a <literal>WHEN NOT MATCHED BY SOURCE</literal> clause, the + expression can only use values from the original row in the target table. + If used in a <literal>WHEN NOT MATCHED [BY TARGET]</literal> clause, the + expression can only use values from the <literal>data_source</literal>. </para> </listitem> </varlistentry> @@ -452,8 +481,9 @@ MERGE <replaceable class="parameter">tot <orderedlist> <listitem> <para> - Evaluate whether each row is <literal>MATCHED</literal> or - <literal>NOT MATCHED</literal>. + Evaluate whether each row is <literal>MATCHED</literal>, + <literal>NOT MATCHED BY SOURCE</literal>, or + <literal>NOT MATCHED [BY TARGET]</literal>. </para> </listitem> <listitem> @@ -528,7 +558,8 @@ MERGE <replaceable class="parameter">tot <para> If a <literal>WHEN</literal> clause omits an <literal>AND</literal> sub-clause, it becomes the final reachable clause of that - kind (<literal>MATCHED</literal> or <literal>NOT MATCHED</literal>). + kind (<literal>MATCHED</literal>, <literal>NOT MATCHED BY SOURCE</literal>, + or <literal>NOT MATCHED [BY TARGET]</literal>). If a later <literal>WHEN</literal> clause of that kind is specified it would be provably unreachable and an error is raised. If no final reachable clause is specified of either kind, it is @@ -619,6 +650,23 @@ WHEN MATCHED THEN temporary table recently loaded into the database. </para> + <para> + Update <literal>wine</literal> based on a replacement wine list, inserting + rows for any new stock, updating modified stock entries, and deleting any + wines not present in the new list. +<programlisting> +MERGE INTO wines w +USING new_wine_list s +ON s.winename = w.winename +WHEN NOT MATCHED BY TARGET THEN + INSERT VALUES(s.winename, s.stock) +WHEN MATCHED AND w.stock != s.stock THEN + UPDATE SET stock = s.stock +WHEN NOT MATCHED BY SOURCE THEN + DELETE; +</programlisting> + </para> + </refsect1> <refsect1> @@ -627,7 +675,10 @@ WHEN MATCHED THEN This command conforms to the <acronym>SQL</acronym> standard. </para> <para> - The WITH clause and <literal>DO NOTHING</literal> action are extensions to + The WITH clause, <literal>BY SOURCE</literal> and + <literal>BY TARGET</literal> qualifiers to + <literal>WHEN NOT MATCHED</literal>, and + <literal>DO NOTHING</literal> action are extensions to the <acronym>SQL</acronym> standard. </para> </refsect1> diff --git a/src/backend/optimizer/prep/prepjointree.c b/src/backend/optimizer/prep/prepjointree.c new file mode 100644 index 4cec12a..ed538e5 --- a/src/backend/optimizer/prep/prepjointree.c +++ b/src/backend/optimizer/prep/prepjointree.c @@ -142,7 +142,6 @@ transform_MERGE_to_join(Query *parse) { RangeTblEntry *joinrte; JoinExpr *joinexpr; - JoinType jointype; int joinrti; List *vars; @@ -152,20 +151,10 @@ transform_MERGE_to_join(Query *parse) /* XXX probably bogus */ vars = NIL; - /* - * When any WHEN NOT MATCHED THEN INSERT clauses exist, we need to use an - * outer join so that we process all unmatched tuples from the source - * relation. If none exist, we can use an inner join. - */ - if (parse->mergeUseOuterJoin) - jointype = JOIN_RIGHT; - else - jointype = JOIN_INNER; - /* Manufacture a join RTE to use. */ joinrte = makeNode(RangeTblEntry); joinrte->rtekind = RTE_JOIN; - joinrte->jointype = jointype; + joinrte->jointype = parse->mergeJoinType; joinrte->joinmergedcols = 0; joinrte->joinaliasvars = vars; joinrte->joinleftcols = NIL; /* MERGE does not allow JOIN USING */ @@ -189,7 +178,7 @@ transform_MERGE_to_join(Query *parse) * Create a JOIN between the target and the source relation. */ joinexpr = makeNode(JoinExpr); - joinexpr->jointype = jointype; + joinexpr->jointype = parse->mergeJoinType; joinexpr->isNatural = false; joinexpr->larg = (Node *) makeNode(RangeTblRef); ((RangeTblRef *) joinexpr->larg)->rtindex = parse->resultRelation; diff --git a/src/backend/optimizer/prep/preptlist.c b/src/backend/optimizer/prep/preptlist.c new file mode 100644 index 137b283..1a4e13e --- a/src/backend/optimizer/prep/preptlist.c +++ b/src/backend/optimizer/prep/preptlist.c @@ -157,15 +157,14 @@ preprocess_targetlist(PlannerInfo *root) /* * Add resjunk entries for any Vars used in each action's * targetlist and WHEN condition that belong to relations other - * than target. Note that aggregates, window functions and - * placeholder vars are not possible anywhere in MERGE's WHEN - * clauses. (PHVs may be added later, but they don't concern us - * here.) + * than target. Note that aggregates and window functions are not + * possible anywhere in MERGE's WHEN clauses, but PlaceHolderVars + * may have been added by subquery pullup. */ vars = pull_var_clause((Node *) list_concat_copy((List *) action->qual, action->targetList), - 0); + PVC_INCLUDE_PLACEHOLDERS); foreach(l2, vars) { Var *var = (Var *) lfirst(l2); diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y new file mode 100644 index 63b4baa..95531ab --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -273,6 +273,7 @@ static Node *makeRecursiveViewSelect(cha struct SelectLimit *selectlimit; SetQuantifier setquantifier; struct GroupClause *groupclause; + MergeMatchKind mergematch; MergeWhenClause *mergewhen; struct KeyActions *keyactions; struct KeyAction *keyaction; @@ -512,6 +513,7 @@ static Node *makeRecursiveViewSelect(cha %type <onconflict> opt_on_conflict %type <mergewhen> merge_insert merge_update merge_delete +%type <mergematch> merge_when_tgt_matched merge_when_tgt_not_matched %type <node> merge_when_clause opt_merge_when_condition %type <list> merge_when_list @@ -740,11 +742,11 @@ static Node *makeRecursiveViewSelect(cha SAVEPOINT SCHEMA SCHEMAS SCROLL SEARCH SECOND_P SECURITY SELECT SEQUENCE SEQUENCES SERIALIZABLE SERVER SESSION SESSION_USER SET SETS SETOF SHARE SHOW - SIMILAR SIMPLE SKIP SMALLINT SNAPSHOT SOME SQL_P STABLE STANDALONE_P + SIMILAR SIMPLE SKIP SMALLINT SNAPSHOT SOME SOURCE SQL_P STABLE STANDALONE_P START STATEMENT STATISTICS STDIN STDOUT STORAGE STORED STRICT_P STRIP_P SUBSCRIPTION SUBSTRING SUPPORT SYMMETRIC SYSID SYSTEM_P SYSTEM_USER - TABLE TABLES TABLESAMPLE TABLESPACE TEMP TEMPLATE TEMPORARY TEXT_P THEN + TABLE TABLES TABLESAMPLE TABLESPACE TARGET TEMP TEMPLATE TEMPORARY TEXT_P THEN TIES TIME TIMESTAMP TO TRAILING TRANSACTION TRANSFORM TREAT TRIGGER TRIM TRUE_P TRUNCATE TRUSTED TYPE_P TYPES_P @@ -12260,49 +12262,59 @@ merge_when_list: ; merge_when_clause: - WHEN MATCHED opt_merge_when_condition THEN merge_update + merge_when_tgt_matched opt_merge_when_condition THEN merge_update { - $5->matched = true; - $5->condition = $3; + $4->matchKind = $1; + $4->condition = $2; - $$ = (Node *) $5; + $$ = (Node *) $4; } - | WHEN MATCHED opt_merge_when_condition THEN merge_delete + | merge_when_tgt_matched opt_merge_when_condition THEN merge_delete { - $5->matched = true; - $5->condition = $3; + $4->matchKind = $1; + $4->condition = $2; - $$ = (Node *) $5; + $$ = (Node *) $4; } - | WHEN NOT MATCHED opt_merge_when_condition THEN merge_insert + | merge_when_tgt_not_matched opt_merge_when_condition THEN merge_insert { - $6->matched = false; - $6->condition = $4; + $4->matchKind = $1; + $4->condition = $2; - $$ = (Node *) $6; + $$ = (Node *) $4; } - | WHEN MATCHED opt_merge_when_condition THEN DO NOTHING + | merge_when_tgt_matched opt_merge_when_condition THEN DO NOTHING { MergeWhenClause *m = makeNode(MergeWhenClause); - m->matched = true; + m->matchKind = $1; m->commandType = CMD_NOTHING; - m->condition = $3; + m->condition = $2; $$ = (Node *) m; } - | WHEN NOT MATCHED opt_merge_when_condition THEN DO NOTHING + | merge_when_tgt_not_matched opt_merge_when_condition THEN DO NOTHING { MergeWhenClause *m = makeNode(MergeWhenClause); - m->matched = false; + m->matchKind = $1; m->commandType = CMD_NOTHING; - m->condition = $4; + m->condition = $2; $$ = (Node *) m; } ; +merge_when_tgt_matched: + WHEN MATCHED { $$ = MERGE_WHEN_MATCHED; } + | WHEN NOT MATCHED BY SOURCE { $$ = MERGE_WHEN_NOT_MATCHED_BY_SOURCE; } + ; + +merge_when_tgt_not_matched: + WHEN NOT MATCHED { $$ = MERGE_WHEN_NOT_MATCHED_BY_TARGET; } + | WHEN NOT MATCHED BY TARGET { $$ = MERGE_WHEN_NOT_MATCHED_BY_TARGET; } + ; + opt_merge_when_condition: AND a_expr { $$ = $2; } | { $$ = NULL; } @@ -16961,6 +16973,7 @@ unreserved_keyword: | SIMPLE | SKIP | SNAPSHOT + | SOURCE | SQL_P | STABLE | STANDALONE_P @@ -16979,6 +16992,7 @@ unreserved_keyword: | SYSTEM_P | TABLES | TABLESPACE + | TARGET | TEMP | TEMPLATE | TEMPORARY @@ -17555,6 +17569,7 @@ bare_label_keyword: | SMALLINT | SNAPSHOT | SOME + | SOURCE | SQL_P | STABLE | STANDALONE_P @@ -17578,6 +17593,7 @@ bare_label_keyword: | TABLES | TABLESAMPLE | TABLESPACE + | TARGET | TEMP | TEMPLATE | TEMPORARY diff --git a/src/backend/parser/parse_merge.c b/src/backend/parser/parse_merge.c new file mode 100644 index 3844f2b..997b7e2 --- a/src/backend/parser/parse_merge.c +++ b/src/backend/parser/parse_merge.c @@ -43,9 +43,9 @@ static void setNamespaceVisibilityForRTE * Make appropriate changes to the namespace visibility while transforming * individual action's quals and targetlist expressions. In particular, for * INSERT actions we must only see the source relation (since INSERT action is - * invoked for NOT MATCHED tuples and hence there is no target tuple to deal - * with). On the other hand, UPDATE and DELETE actions can see both source and - * target relations. + * invoked for NOT MATCHED [BY TARGET] tuples and hence there is no target + * tuple to deal with). On the other hand, UPDATE and DELETE actions can see + * both source and target relations, unless invoked for NOT MATCH BY SOURCE. * * Also, since the internal join node can hide the source and target * relations, we must explicitly make the respective relation as visible so @@ -61,7 +61,7 @@ setNamespaceForMergeWhen(ParseState *pst targetRelRTE = rt_fetch(targetRTI, pstate->p_rtable); sourceRelRTE = rt_fetch(sourceRTI, pstate->p_rtable); - if (mergeWhenClause->matched) + if (mergeWhenClause->matchKind == MERGE_WHEN_MATCHED) { Assert(mergeWhenClause->commandType == CMD_UPDATE || mergeWhenClause->commandType == CMD_DELETE || @@ -73,11 +73,25 @@ setNamespaceForMergeWhen(ParseState *pst setNamespaceVisibilityForRTE(pstate->p_namespace, sourceRelRTE, true, true); } - else + else if (mergeWhenClause->matchKind == MERGE_WHEN_NOT_MATCHED_BY_SOURCE) { /* - * NOT MATCHED actions can't see target relation, but they can see - * source relation. + * NOT MATCHED BY SOURCE actions can see the target relation, but they + * can't see the source relation. + */ + Assert(mergeWhenClause->commandType == CMD_UPDATE || + mergeWhenClause->commandType == CMD_DELETE || + mergeWhenClause->commandType == CMD_NOTHING); + setNamespaceVisibilityForRTE(pstate->p_namespace, + targetRelRTE, true, true); + setNamespaceVisibilityForRTE(pstate->p_namespace, + sourceRelRTE, false, false); + } + else /* mergeWhenClause->matchKind == MERGE_WHEN_NOT_MATCHED_BY_TARGET */ + { + /* + * NOT MATCHED [BY TARGET] actions can't see target relation, but they + * can see source relation. */ Assert(mergeWhenClause->commandType == CMD_INSERT || mergeWhenClause->commandType == CMD_NOTHING); @@ -98,7 +112,9 @@ transformMergeStmt(ParseState *pstate, M Query *qry = makeNode(Query); ListCell *l; AclMode targetPerms = ACL_NO_RIGHTS; - bool is_terminal[2]; + bool is_terminal[3]; + bool left_join; + bool right_join; Index sourceRTI; List *mergeActionList; Node *joinExpr; @@ -127,10 +143,12 @@ transformMergeStmt(ParseState *pstate, M */ is_terminal[0] = false; is_terminal[1] = false; + is_terminal[2] = false; + left_join = false; + right_join = false; foreach(l, stmt->mergeWhenClauses) { MergeWhenClause *mergeWhenClause = (MergeWhenClause *) lfirst(l); - int when_type = (mergeWhenClause->matched ? 0 : 1); /* * Collect action types so we can check target permissions @@ -155,14 +173,33 @@ transformMergeStmt(ParseState *pstate, M /* * Check for unreachable WHEN clauses */ - if (mergeWhenClause->condition == NULL) - is_terminal[when_type] = true; - else if (is_terminal[when_type]) + if (is_terminal[mergeWhenClause->matchKind]) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("unreachable WHEN clause specified after unconditional WHEN clause"))); + if (mergeWhenClause->condition == NULL) + is_terminal[mergeWhenClause->matchKind] = true; + + /* Is left/right/full outer join required? */ + if (mergeWhenClause->commandType != CMD_NOTHING) + { + if (mergeWhenClause->matchKind == MERGE_WHEN_NOT_MATCHED_BY_SOURCE) + left_join = true; + if (mergeWhenClause->matchKind == MERGE_WHEN_NOT_MATCHED_BY_TARGET) + right_join = true; + } } + /* Join type required */ + if (left_join && right_join) + qry->mergeJoinType = JOIN_FULL; + else if (left_join) + qry->mergeJoinType = JOIN_LEFT; + else if (right_join) + qry->mergeJoinType = JOIN_RIGHT; + else + qry->mergeJoinType = JOIN_INNER; + /* * Set up the MERGE target table. The target table is added to the * namespace below and to joinlist in transform_MERGE_to_join, so don't @@ -251,15 +288,19 @@ transformMergeStmt(ParseState *pstate, M foreach(l, stmt->mergeWhenClauses) { MergeWhenClause *mergeWhenClause = lfirst_node(MergeWhenClause, l); + MergeMatchKind matchKind = mergeWhenClause->matchKind; MergeAction *action; action = makeNode(MergeAction); action->commandType = mergeWhenClause->commandType; - action->matched = mergeWhenClause->matched; - /* Use an outer join if any INSERT actions exist in the command. */ - if (action->commandType == CMD_INSERT) - qry->mergeUseOuterJoin = true; + /* + * For planning and execution purposes, MATCHED and NOT MATCHED BY + * SOURCE actions are handled the same, since they each must match + * target rows. Thus we need only note whether the action is NOT + * MATCHED [BY TARGET]. + */ + action->matched = matchKind != MERGE_WHEN_NOT_MATCHED_BY_TARGET; /* * Set namespace for the specific action. This must be done before @@ -280,6 +321,50 @@ transformMergeStmt(ParseState *pstate, M EXPR_KIND_MERGE_WHEN, "WHEN"); /* + * If we are doing a left or full join that includes target-only rows + * (NOT MATCHED BY SOURCE), add additional WHEN conditions to each + * action to test if the source row matches. + * + * This distinguishes NOT MATCHED BY SOURCE actions (identified by a + * "source IS NOT DISTINCT FROM NULL" clause) from MATCHED actions + * (identified by a "source IS DISTINCT FROM NULL" clause). + * + * Additionally, a "source IS DISTINCT FROM NULL" clause is required + * for NOT MATCHED [BY TARGET] actions in case the executor attempts + * to invoke such an action for a concurrently deleted target row that + * ends up matching neither source nor target. + */ + if (left_join) + { + bool src_null; + NullTest *ntest; + + /* Make a "source IS [NOT] DISTINCT FROM NULL" clause */ + src_null = matchKind == MERGE_WHEN_NOT_MATCHED_BY_SOURCE; + + ntest = makeNode(NullTest); + + ntest->arg = (Expr *) makeVar(sourceRTI, + InvalidAttrNumber, + RECORDOID, + -1, + InvalidOid, + 0); + ntest->nulltesttype = src_null ? IS_NULL : IS_NOT_NULL; + ntest->argisrow = false; + ntest->location = -1; + + /* Combine it with the action's WHEN condition */ + if (action->qual == NULL) + action->qual = (Node *) ntest; + else + action->qual = + (Node *) makeBoolExpr(AND_EXPR, + list_make2(ntest, action->qual), + -1); + } + + /* * Transform target lists for each INSERT and UPDATE action stmt */ switch (action->commandType) diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h new file mode 100644 index 34bc640..e9636c8 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -159,7 +159,7 @@ typedef struct Query * also USING clause for MERGE */ List *mergeActionList; /* list of actions for MERGE (only) */ - bool mergeUseOuterJoin; /* whether to use outer join */ + JoinType mergeJoinType; /* MERGE target to source join type */ List *targetList; /* target list (of TargetEntry) */ @@ -1586,6 +1586,14 @@ typedef struct CommonTableExpr ((Query *) (cte)->ctequery)->targetList : \ ((Query *) (cte)->ctequery)->returningList) +/* MATCHED/NOT MATCHED specification for MergeWhenClause */ +typedef enum MergeMatchKind +{ + MERGE_WHEN_MATCHED, + MERGE_WHEN_NOT_MATCHED_BY_SOURCE, + MERGE_WHEN_NOT_MATCHED_BY_TARGET +} MergeMatchKind; + /* * MergeWhenClause - * raw parser representation of a WHEN clause in a MERGE statement @@ -1595,7 +1603,7 @@ typedef struct CommonTableExpr typedef struct MergeWhenClause { NodeTag type; - bool matched; /* true=MATCHED, false=NOT MATCHED */ + MergeMatchKind matchKind; /* MATCHED/NOT MATCHED BY SOURCE/TARGET */ CmdType commandType; /* INSERT/UPDATE/DELETE/DO NOTHING */ OverridingKind override; /* OVERRIDING clause */ Node *condition; /* WHEN conditions (raw parser) */ @@ -1611,7 +1619,8 @@ typedef struct MergeWhenClause typedef struct MergeAction { NodeTag type; - bool matched; /* true=MATCHED, false=NOT MATCHED */ + bool matched; /* true if target matches (input matchKind is + * MATCHED or NOT MATCHED BY SOURCE) */ CmdType commandType; /* INSERT/UPDATE/DELETE/DO NOTHING */ OverridingKind override; /* OVERRIDING clause */ Node *qual; /* transformed WHEN conditions */ diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h new file mode 100644 index 957ee18..fd94b0b --- a/src/include/parser/kwlist.h +++ b/src/include/parser/kwlist.h @@ -391,6 +391,7 @@ PG_KEYWORD("skip", SKIP, UNRESERVED_KEYW PG_KEYWORD("smallint", SMALLINT, COL_NAME_KEYWORD, BARE_LABEL) PG_KEYWORD("snapshot", SNAPSHOT, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("some", SOME, RESERVED_KEYWORD, BARE_LABEL) +PG_KEYWORD("source", SOURCE, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("sql", SQL_P, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("stable", STABLE, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("standalone", STANDALONE_P, UNRESERVED_KEYWORD, BARE_LABEL) @@ -414,6 +415,7 @@ PG_KEYWORD("table", TABLE, RESERVED_KEYW PG_KEYWORD("tables", TABLES, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("tablesample", TABLESAMPLE, TYPE_FUNC_NAME_KEYWORD, BARE_LABEL) PG_KEYWORD("tablespace", TABLESPACE, UNRESERVED_KEYWORD, BARE_LABEL) +PG_KEYWORD("target", TARGET, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("temp", TEMP, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("template", TEMPLATE, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("temporary", TEMPORARY, UNRESERVED_KEYWORD, BARE_LABEL) diff --git a/src/test/regress/expected/merge.out b/src/test/regress/expected/merge.out new file mode 100644 index 6c8a18f..e99623a --- a/src/test/regress/expected/merge.out +++ b/src/test/regress/expected/merge.out @@ -71,6 +71,15 @@ WHEN MATCHED THEN ERROR: syntax error at or near "INSERT" LINE 5: INSERT DEFAULT VALUES; ^ +-- NOT MATCHED BY SOURCE/INSERT error +MERGE INTO target t +USING source AS s +ON t.tid = s.sid +WHEN NOT MATCHED BY SOURCE THEN + INSERT DEFAULT VALUES; +ERROR: syntax error at or near "INSERT" +LINE 5: INSERT DEFAULT VALUES; + ^ -- incorrectly specifying INTO target MERGE INTO target t USING source AS s @@ -107,6 +116,15 @@ WHEN NOT MATCHED THEN ERROR: syntax error at or near "UPDATE" LINE 5: UPDATE SET balance = 0; ^ +-- NOT MATCHED BY TARGET/UPDATE +MERGE INTO target t +USING source AS s +ON t.tid = s.sid +WHEN NOT MATCHED BY TARGET THEN + UPDATE SET balance = 0; +ERROR: syntax error at or near "UPDATE" +LINE 5: UPDATE SET balance = 0; + ^ -- UPDATE tablename MERGE INTO target t USING source AS s @@ -271,6 +289,22 @@ SELECT * FROM target ORDER BY tid; (4 rows) ROLLBACK; +-- DELETE/INSERT not matched by source/target +BEGIN; +MERGE INTO target t +USING source AS s +ON t.tid = s.sid +WHEN NOT MATCHED BY SOURCE THEN + DELETE +WHEN NOT MATCHED BY TARGET THEN + INSERT VALUES (s.sid, s.delta); +SELECT * FROM target ORDER BY tid; + tid | balance +-----+--------- + 4 | 40 +(1 row) + +ROLLBACK; -- index plans INSERT INTO target SELECT generate_series(1000,2500), 0; ALTER TABLE target ADD PRIMARY KEY (tid); @@ -738,6 +772,19 @@ SELECT * FROM wq_target; 1 | -1 (1 row) +-- conditions in NOT MATCHED BY SOURCE clause can only refer to target columns +MERGE INTO wq_target t +USING wq_source s ON t.tid = s.sid +WHEN NOT MATCHED BY SOURCE AND s.balance = 100 THEN + DELETE; +ERROR: invalid reference to FROM-clause entry for table "s" +LINE 3: WHEN NOT MATCHED BY SOURCE AND s.balance = 100 THEN + ^ +DETAIL: There is an entry for table "s", but it cannot be referenced from this part of the query. +MERGE INTO wq_target t +USING wq_source s ON t.tid = s.sid +WHEN NOT MATCHED BY SOURCE AND t.balance = 100 THEN + DELETE; -- conditions in MATCHED clause can refer to both source and target SELECT * FROM wq_source; balance | sid @@ -923,6 +970,45 @@ SELECT * FROM target ORDER BY tid; (3 rows) ROLLBACK; +-- UPSERT with UPDATE/DELETE not matched +BEGIN; +DELETE FROM SOURCE WHERE sid = 2; +MERGE INTO target t +USING source AS s +ON t.tid = s.sid +WHEN MATCHED AND t.balance > s.delta THEN + UPDATE SET balance = t.balance - s.delta +WHEN MATCHED THEN + UPDATE SET balance = 0 +WHEN NOT MATCHED THEN + INSERT VALUES (s.sid, s.delta) +WHEN NOT MATCHED BY SOURCE AND tid = 1 THEN + UPDATE SET balance = 0 +WHEN NOT MATCHED BY SOURCE THEN + DELETE; +NOTICE: BEFORE INSERT STATEMENT trigger +NOTICE: BEFORE UPDATE STATEMENT trigger +NOTICE: BEFORE DELETE STATEMENT trigger +NOTICE: BEFORE UPDATE ROW trigger row: (3,30) -> (3,10) +NOTICE: BEFORE INSERT ROW trigger row: (4,40) +NOTICE: BEFORE DELETE ROW trigger row: (2,20) +NOTICE: BEFORE UPDATE ROW trigger row: (1,10) -> (1,0) +NOTICE: AFTER UPDATE ROW trigger row: (3,30) -> (3,10) +NOTICE: AFTER INSERT ROW trigger row: (4,40) +NOTICE: AFTER DELETE ROW trigger row: (2,20) +NOTICE: AFTER UPDATE ROW trigger row: (1,10) -> (1,0) +NOTICE: AFTER DELETE STATEMENT trigger +NOTICE: AFTER UPDATE STATEMENT trigger +NOTICE: AFTER INSERT STATEMENT trigger +SELECT * FROM target ORDER BY tid; + tid | balance +-----+--------- + 1 | 0 + 3 | 10 + 4 | 40 +(3 rows) + +ROLLBACK; -- Test behavior of triggers that turn UPDATE/DELETE into no-ops create or replace function skip_merge_op() returns trigger language plpgsql as @@ -1436,6 +1522,50 @@ WHEN NOT MATCHED AND s.a < 20 THEN -> Seq Scan on ex_mtarget t (actual rows=49 loops=1) (12 rows) +-- not matched by source +SELECT explain_merge(' +MERGE INTO ex_mtarget t USING ex_msource s ON t.a = s.a +WHEN NOT MATCHED BY SOURCE and t.a < 10 THEN + DELETE'); + explain_merge +---------------------------------------------------------------------- + Merge on ex_mtarget t (actual rows=0 loops=1) + Tuples: skipped=54 + -> Merge Left Join (actual rows=54 loops=1) + Merge Cond: (t.a = s.a) + -> Sort (actual rows=54 loops=1) + Sort Key: t.a + Sort Method: quicksort Memory: xxx + -> Seq Scan on ex_mtarget t (actual rows=54 loops=1) + -> Sort (actual rows=100 loops=1) + Sort Key: s.a + Sort Method: quicksort Memory: xxx + -> Seq Scan on ex_msource s (actual rows=100 loops=1) +(12 rows) + +-- not matched by source and target +SELECT explain_merge(' +MERGE INTO ex_mtarget t USING ex_msource s ON t.a = s.a +WHEN NOT MATCHED BY SOURCE AND t.a < 10 THEN + DELETE +WHEN NOT MATCHED BY TARGET AND s.a < 20 THEN + INSERT VALUES (a, b)'); + explain_merge +---------------------------------------------------------------------- + Merge on ex_mtarget t (actual rows=0 loops=1) + Tuples: skipped=100 + -> Merge Full Join (actual rows=100 loops=1) + Merge Cond: (t.a = s.a) + -> Sort (actual rows=54 loops=1) + Sort Key: t.a + Sort Method: quicksort Memory: xxx + -> Seq Scan on ex_mtarget t (actual rows=54 loops=1) + -> Sort (actual rows=100 loops=1) + Sort Key: s.a + Sort Method: quicksort Memory: xxx + -> Seq Scan on ex_msource s (actual rows=100 loops=1) +(12 rows) + -- nothing SELECT explain_merge(' MERGE INTO ex_mtarget t USING ex_msource s ON t.a = s.a AND t.a < -1000 diff --git a/src/test/regress/sql/merge.sql b/src/test/regress/sql/merge.sql new file mode 100644 index 98fe104..c641214 --- a/src/test/regress/sql/merge.sql +++ b/src/test/regress/sql/merge.sql @@ -51,6 +51,12 @@ USING source AS s ON t.tid = s.sid WHEN MATCHED THEN INSERT DEFAULT VALUES; +-- NOT MATCHED BY SOURCE/INSERT error +MERGE INTO target t +USING source AS s +ON t.tid = s.sid +WHEN NOT MATCHED BY SOURCE THEN + INSERT DEFAULT VALUES; -- incorrectly specifying INTO target MERGE INTO target t USING source AS s @@ -75,6 +81,12 @@ USING source AS s ON t.tid = s.sid WHEN NOT MATCHED THEN UPDATE SET balance = 0; +-- NOT MATCHED BY TARGET/UPDATE +MERGE INTO target t +USING source AS s +ON t.tid = s.sid +WHEN NOT MATCHED BY TARGET THEN + UPDATE SET balance = 0; -- UPDATE tablename MERGE INTO target t USING source AS s @@ -213,6 +225,18 @@ WHEN NOT MATCHED THEN SELECT * FROM target ORDER BY tid; ROLLBACK; +-- DELETE/INSERT not matched by source/target +BEGIN; +MERGE INTO target t +USING source AS s +ON t.tid = s.sid +WHEN NOT MATCHED BY SOURCE THEN + DELETE +WHEN NOT MATCHED BY TARGET THEN + INSERT VALUES (s.sid, s.delta); +SELECT * FROM target ORDER BY tid; +ROLLBACK; + -- index plans INSERT INTO target SELECT generate_series(1000,2500), 0; ALTER TABLE target ADD PRIMARY KEY (tid); @@ -497,6 +521,17 @@ WHEN NOT MATCHED AND s.balance = 100 THE INSERT (tid) VALUES (s.sid); SELECT * FROM wq_target; +-- conditions in NOT MATCHED BY SOURCE clause can only refer to target columns +MERGE INTO wq_target t +USING wq_source s ON t.tid = s.sid +WHEN NOT MATCHED BY SOURCE AND s.balance = 100 THEN + DELETE; + +MERGE INTO wq_target t +USING wq_source s ON t.tid = s.sid +WHEN NOT MATCHED BY SOURCE AND t.balance = 100 THEN + DELETE; + -- conditions in MATCHED clause can refer to both source and target SELECT * FROM wq_source; MERGE INTO wq_target t @@ -624,6 +659,25 @@ WHEN NOT MATCHED THEN SELECT * FROM target ORDER BY tid; ROLLBACK; +-- UPSERT with UPDATE/DELETE not matched +BEGIN; +DELETE FROM SOURCE WHERE sid = 2; +MERGE INTO target t +USING source AS s +ON t.tid = s.sid +WHEN MATCHED AND t.balance > s.delta THEN + UPDATE SET balance = t.balance - s.delta +WHEN MATCHED THEN + UPDATE SET balance = 0 +WHEN NOT MATCHED THEN + INSERT VALUES (s.sid, s.delta) +WHEN NOT MATCHED BY SOURCE AND tid = 1 THEN + UPDATE SET balance = 0 +WHEN NOT MATCHED BY SOURCE THEN + DELETE; +SELECT * FROM target ORDER BY tid; +ROLLBACK; + -- Test behavior of triggers that turn UPDATE/DELETE into no-ops create or replace function skip_merge_op() returns trigger language plpgsql as @@ -916,6 +970,20 @@ WHEN MATCHED AND t.a >= 30 AND t.a <= 40 WHEN NOT MATCHED AND s.a < 20 THEN INSERT VALUES (a, b)'); +-- not matched by source +SELECT explain_merge(' +MERGE INTO ex_mtarget t USING ex_msource s ON t.a = s.a +WHEN NOT MATCHED BY SOURCE and t.a < 10 THEN + DELETE'); + +-- not matched by source and target +SELECT explain_merge(' +MERGE INTO ex_mtarget t USING ex_msource s ON t.a = s.a +WHEN NOT MATCHED BY SOURCE AND t.a < 10 THEN + DELETE +WHEN NOT MATCHED BY TARGET AND s.a < 20 THEN + INSERT VALUES (a, b)'); + -- nothing SELECT explain_merge(' MERGE INTO ex_mtarget t USING ex_msource s ON t.a = s.a AND t.a < -1000 ^ permalink raw reply [nested|flat] 2+ messages in thread
end of thread, other threads:[~2023-01-02 18:24 UTC | newest] Thread overview: 2+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-02-10 06:19 [PATCH v24 7/7] Add support for PRESERVE Dilip Kumar <[email protected]> 2023-01-02 18:24 Re: MERGE ... WHEN NOT MATCHED BY SOURCE Dean Rasheed <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox