PostgreSQL create or replace view和重建视图 有什么区别？

一、 replace vs 重建

       遇到开发提了个问题，create or replace view和重建视图（drop+create）有什么区别，查询资料整理了一下。

1. create or replace

• 当存在同名视图时，尝试将其替换
• 新视图语句必须与现有视图查询具有相同的列（即相同的列名、列顺序和数据类型）
• pg 8.1开始，之前向视图末尾添加新列
• 总体而言，改动限制较大，但replace后不影响权限和依赖于该视图的对象

2. drop+create

• 新视图定义不依赖原视图，灵活度高
• 重建后权限丢失，必须重新授权

-- 对象授权查询
SELECT * FROM information_schema.table_privileges WHERE table_name='视图名' and grantor<>grantee;

• 依赖于该视图的对象需要一起重建，复杂度可能较高

二、 源码学习

1. CREATE OR REPLACE VIEW

       按照 "CREATE OR REPLACE VIEW" 关键字搜索，这部分代码在ATExecCmd函数（tablecmds.c文件）。可以看到它对应的命令类型叫AT_AddColumnToView，对应操作为调用ATExecAddColumn函数为视图新加列。

/** ATExecCmd: dispatch a subcommand to appropriate execution routine*/
static void
ATExecCmd(List **wqueue, AlteredTableInfo *tab,AlterTableCmd *cmd, LOCKMODE lockmode, int cur_pass,AlterTableUtilityContext *context)
{ObjectAddress address = InvalidObjectAddress;Relation	rel = tab->rel;switch (cmd->subtype){case AT_AddColumn:		/* ADD COLUMN */case AT_AddColumnToView:	/* add column via CREATE OR REPLACE VIEW */address = ATExecAddColumn(wqueue, tab, rel, &cmd,false, false,lockmode, cur_pass, context);break;
...

2. ATExecAddColumn函数

比较长的一个函数，主要操作如下：

• 权限检查

/** Add a column to a table.  The return value is the address of the* new column in the parent relation.** cmd is pass-by-ref so that we can replace it with the parse-transformed* copy (but that happens only after we check for IF NOT EXISTS).*/
static ObjectAddress
ATExecAddColumn(List **wqueue, AlteredTableInfo *tab, Relation rel,AlterTableCmd **cmd,bool recurse, bool recursing,LOCKMODE lockmode, int cur_pass,AlterTableUtilityContext *context)
{Oid			myrelid = RelationGetRelid(rel);ColumnDef  *colDef = castNode(ColumnDef, (*cmd)->def);bool		if_not_exists = (*cmd)->missing_ok;Relation	pgclass,attrdesc;
.../* At top level, permission check was done in ATPrepCmd, else do it */if (recursing)ATSimplePermissions(rel, ATT_TABLE | ATT_FOREIGN_TABLE);if (rel->rd_rel->relispartition && !recursing)ereport(ERROR,(errcode(ERRCODE_WRONG_OBJECT_TYPE),errmsg("cannot add column to a partition")));

• 获取表锁，等级为RowExclusiveLock

attrdesc = table_open(AttributeRelationId, RowExclusiveLock);

• 判断是否为递归子表加列，能否进行merge列

  若是merge，要求更新前后列类型及排序规则一致（Child column must match on type, typmod, and collation）

	/** Are we adding the column to a recursion child?  If so, check whether to* merge with an existing definition for the column.  If we do merge, we* must not recurse.  Children will already have the column, and recursing* into them would mess up attinhcount.*/if (colDef->inhcount > 0){HeapTuple	tuple;/* Does child already have a column by this name? */tuple = SearchSysCacheCopyAttName(myrelid, colDef->colname);if (HeapTupleIsValid(tuple)){Form_pg_attribute childatt = (Form_pg_attribute) GETSTRUCT(tuple);Oid			ctypeId;int32		ctypmod;Oid			ccollid;/* Child column must match on type, typmod, and collation */typenameTypeIdAndMod(NULL, colDef->typeName, &ctypeId, &ctypmod);if (ctypeId != childatt->atttypid ||ctypmod != childatt->atttypmod)ereport(ERROR,(errcode(ERRCODE_DATATYPE_MISMATCH),errmsg("child table \"%s\" has different type for column \"%s\"",RelationGetRelationName(rel), colDef->colname)));ccollid = GetColumnDefCollation(NULL, colDef, ctypeId);if (ccollid != childatt->attcollation)ereport(ERROR,(errcode(ERRCODE_COLLATION_MISMATCH),errmsg("child table \"%s\" has different collation for column \"%s\"",RelationGetRelationName(rel), colDef->colname),errdetail("\"%s\" versus \"%s\"",get_collation_name(ccollid),get_collation_name(childatt->attcollation))));/* Bump the existing child att's inhcount */childatt->attinhcount++;CatalogTupleUpdate(attrdesc, &tuple->t_self, tuple);heap_freetuple(tuple);/* Inform the user about the merge */ereport(NOTICE,(errmsg("merging definition of column \"%s\" for child \"%s\"",colDef->colname, RelationGetRelationName(rel))));table_close(attrdesc, RowExclusiveLock);return InvalidObjectAddress;}}

• 一些检查，例如列名是否已存在，列数是否超出限制等

	/* skip if the name already exists and if_not_exists is true */if (!check_for_column_name_collision(rel, colDef->colname, if_not_exists)){table_close(attrdesc, RowExclusiveLock);return InvalidObjectAddress;}
.../* Determine the new attribute's number */newattnum = ((Form_pg_class) GETSTRUCT(reltup))->relnatts + 1;if (newattnum > MaxHeapAttributeNumber)ereport(ERROR,(errcode(ERRCODE_TOO_MANY_COLUMNS),errmsg("tables can have at most %d columns",MaxHeapAttributeNumber)));

• 填充pg_attribute及pg_class信息

	/* construct new attribute's pg_attribute entry */attribute.attrelid = myrelid;namestrcpy(&(attribute.attname), colDef->colname);attribute.atttypid = typeOid;attribute.attstattarget = (newattnum > 0) ? -1 : 0;attribute.attlen = tform->typlen;attribute.attnum = newattnum;attribute.attndims = list_length(colDef->typeName->arrayBounds);attribute.atttypmod = typmod;attribute.attbyval = tform->typbyval;attribute.attalign = tform->typalign;attribute.attstorage = tform->typstorage;attribute.attcompression = GetAttributeCompression(typeOid,colDef->compression);attribute.attnotnull = colDef->is_not_null;attribute.atthasdef = false;attribute.atthasmissing = false;attribute.attidentity = colDef->identity;attribute.attgenerated = colDef->generated;attribute.attisdropped = false;attribute.attislocal = colDef->is_local;attribute.attinhcount = colDef->inhcount;attribute.attcollation = collOid;/* attribute.attacl is handled by InsertPgAttributeTuples() */ReleaseSysCache(typeTuple);tupdesc = CreateTupleDesc(lengthof(aattr), (FormData_pg_attribute **) &aattr);InsertPgAttributeTuples(attrdesc, tupdesc, myrelid, NULL, NULL);table_close(attrdesc, RowExclusiveLock);/** Update pg_class tuple as appropriate*/((Form_pg_class) GETSTRUCT(reltup))->relnatts = newattnum;CatalogTupleUpdate(pgclass, &reltup->t_self, reltup);heap_freetuple(reltup);/* Post creation hook for new attribute */InvokeObjectPostCreateHook(RelationRelationId, myrelid, newattnum);table_close(pgclass, RowExclusiveLock);/* Make the attribute's catalog entry visible */CommandCounterIncrement();

• 填充默认值

/** Store the DEFAULT, if any, in the catalogs*/if (colDef->raw_default){RawColumnDefault *rawEnt;rawEnt = (RawColumnDefault *) palloc(sizeof(RawColumnDefault));rawEnt->attnum = attribute.attnum;rawEnt->raw_default = copyObject(colDef->raw_default);/** Attempt to skip a complete table rewrite by storing the specified* DEFAULT value outside of the heap.  This may be disabled inside* AddRelationNewConstraints if the optimization cannot be applied.*/rawEnt->missingMode = (!colDef->generated);rawEnt->generated = colDef->generated;/** This function is intended for CREATE TABLE, so it processes a* _list_ of defaults, but we just do one.*/AddRelationNewConstraints(rel, list_make1(rawEnt), NIL,false, true, false, NULL);/* Make the additional catalog changes visible */CommandCounterIncrement();/** Did the request for a missing value work? If not we'll have to do a* rewrite*/if (!rawEnt->missingMode)tab->rewrite |= AT_REWRITE_DEFAULT_VAL;}/** Tell Phase 3 to fill in the default expression, if there is one.** If there is no default, Phase 3 doesn't have to do anything, because* that effectively means that the default is NULL.  The heap tuple access* routines always check for attnum > # of attributes in tuple, and return* NULL if so, so without any modification of the tuple data we will get* the effect of NULL values in the new column.** An exception occurs when the new column is of a domain type: the domain* might have a NOT NULL constraint, or a check constraint that indirectly* rejects nulls.  If there are any domain constraints then we construct* an explicit NULL default value that will be passed through* CoerceToDomain processing.  (This is a tad inefficient, since it causes* rewriting the table which we really don't have to do, but the present* design of domain processing doesn't offer any simple way of checking* the constraints more directly.)** Note: we use build_column_default, and not just the cooked default* returned by AddRelationNewConstraints, so that the right thing happens* when a datatype's default applies.** Note: it might seem that this should happen at the end of Phase 2, so* that the effects of subsequent subcommands can be taken into account.* It's intentional that we do it now, though.  The new column should be* filled according to what is said in the ADD COLUMN subcommand, so that* the effects are the same as if this subcommand had been run by itself* and the later subcommands had been issued in new ALTER TABLE commands.** We can skip this entirely for relations without storage, since Phase 3* is certainly not going to touch them.  System attributes don't have* interesting defaults, either.*/if (RELKIND_HAS_STORAGE(relkind) && attribute.attnum > 0){/** For an identity column, we can't use build_column_default(),* because the sequence ownership isn't set yet.  So do it manually.*/if (colDef->identity){NextValueExpr *nve = makeNode(NextValueExpr);nve->seqid = RangeVarGetRelid(colDef->identitySequence, NoLock, false);nve->typeId = typeOid;defval = (Expr *) nve;/* must do a rewrite for identity columns */tab->rewrite |= AT_REWRITE_DEFAULT_VAL;}elsedefval = (Expr *) build_column_default(rel, attribute.attnum);if (!defval && DomainHasConstraints(typeOid)){Oid			baseTypeId;int32		baseTypeMod;Oid			baseTypeColl;baseTypeMod = typmod;baseTypeId = getBaseTypeAndTypmod(typeOid, &baseTypeMod);baseTypeColl = get_typcollation(baseTypeId);defval = (Expr *) makeNullConst(baseTypeId, baseTypeMod, baseTypeColl);defval = (Expr *) coerce_to_target_type(NULL,(Node *) defval,baseTypeId,typeOid,typmod,COERCION_ASSIGNMENT,COERCE_IMPLICIT_CAST,-1);if (defval == NULL) /* should not happen */elog(ERROR, "failed to coerce base type to domain");}if (defval){NewColumnValue *newval;newval = (NewColumnValue *) palloc0(sizeof(NewColumnValue));newval->attnum = attribute.attnum;newval->expr = expression_planner(defval);newval->is_generated = (colDef->generated != '\0');tab->newvals = lappend(tab->newvals, newval);}if (DomainHasConstraints(typeOid))tab->rewrite |= AT_REWRITE_DEFAULT_VAL;if (!TupleDescAttr(rel->rd_att, attribute.attnum - 1)->atthasmissing){/** If the new column is NOT NULL, and there is no missing value,* tell Phase 3 it needs to check for NULLs.*/tab->verify_new_notnull |= colDef->is_not_null;}}

• 更新pg_attribute的atttypid及attcollation字段

	/** Add needed dependency entries for the new column.*/add_column_datatype_dependency(myrelid, newattnum, attribute.atttypid);add_column_collation_dependency(myrelid, newattnum, attribute.attcollation);

• 若有子表，为每个子表递归调用该函数新增列

/** Propagate to children as appropriate.  Unlike most other ALTER* routines, we have to do this one level of recursion at a time; we can't* use find_all_inheritors to do it in one pass.*/children =find_inheritance_children(RelationGetRelid(rel), lockmode);/** If we are told not to recurse, there had better not be any child* tables; else the addition would put them out of step.*/if (children && !recurse)ereport(ERROR,(errcode(ERRCODE_INVALID_TABLE_DEFINITION),errmsg("column must be added to child tables too")));/* Children should see column as singly inherited */if (!recursing){childcmd = copyObject(*cmd);colDef = castNode(ColumnDef, childcmd->def);colDef->inhcount = 1;colDef->is_local = false;}elsechildcmd = *cmd;		/* no need to copy again */foreach(child, children){Oid			childrelid = lfirst_oid(child);Relation	childrel;AlteredTableInfo *childtab;/* find_inheritance_children already got lock */childrel = table_open(childrelid, NoLock);CheckTableNotInUse(childrel, "ALTER TABLE");/* Find or create work queue entry for this table */childtab = ATGetQueueEntry(wqueue, childrel);/* Recurse to child; return value is ignored */ATExecAddColumn(wqueue, childtab, childrel,&childcmd, recurse, true,lockmode, cur_pass, context);table_close(childrel, NoLock);}ObjectAddressSubSet(address, RelationRelationId, myrelid, newattnum);return address;
}

参考

PostgreSQL: Documentation: 16: CREATE VIEW