pgsql-fix-inherit-rename.7.patch

*** a/src/backend/catalog/pg_inherits.c
--- b/src/backend/catalog/pg_inherits.c
***************
*** 148,163 **** find_inheritance_children(Oid parentrelId, LOCKMODE lockmode)
   * find_all_inheritors -
   *		Returns a list of relation OIDs including the given rel plus
   *		all relations that inherit from it, directly or indirectly.
   *
   * The specified lock type is acquired on all child relations (but not on the
   * given rel; caller should already have locked it).  If lockmode is NoLock
   * then no locks are acquired, but caller must beware of race conditions
   * against possible DROPs of child relations.
   */
! List *
! find_all_inheritors(Oid parentrelId, LOCKMODE lockmode)
  {
! 	List	   *rels_list;
  	ListCell   *l;
  
  	/*
--- 148,194 ----
   * find_all_inheritors -
   *		Returns a list of relation OIDs including the given rel plus
   *		all relations that inherit from it, directly or indirectly.
+  *		In addition, it also returns the expected inhcount for them.
   *
   * The specified lock type is acquired on all child relations (but not on the
   * given rel; caller should already have locked it).  If lockmode is NoLock
   * then no locks are acquired, but caller must beware of race conditions
   * against possible DROPs of child relations.
+  *
+  * The expected inhcount means the number of times a certain relation being
+  * inherited from any other relations which were inherited from the root
+  * of inheritance tree directly or indirectly.
+  *
+  * It is used to control some of ALTER TABLE operations on columns inherited
+  * from a relation. For example, an inherited column must has same name
+  * within the series of inheritance. So, we need to prevent renaming when
+  * the column to be renamed is delivered from multiple independent relations.
+  *
+  * The specified lock type is acquired on all child relations (but not on the
+  * given rel; caller should already have locked it).  If lockmode is NoLock
+  * then no locks are acquired, but caller must beware of race conditions
+  * against possible DROPs of child relations.
+  *
+  *    T2     For example, T2 and T3 are inherited from T1, T4 is inherited
+  *   /  \    from T2 and T3. And, T5 is inherited from T3 and S1.
+  * T1    T4  When we try to rename a column within T1, it needs to rename
+  *   \  /    inherited column in the child relations also.
+  *    T3-T5  If attinhcount of the column is larger than its expected inhcount,
+  *      /    it means the column is delivered from the different origin, so
+  *    S1     we have to prevent this operations to keep database integrity.
+  *
+  * In this example, the expected inhcount is 1 in T2, T3 and T5, and 2 in T4.
+  * The column inherited from T1 has attinhcount = 2 in T4, so it does not
+  * prevent renaming.
+  *
+  * If the column inherited from T1 has attinhcount = 2 in T5, it means the
+  * column was also inherited from S1, so we can prevent renaming it.
   */
! void
! find_all_inheritors(Oid parentrelId, LOCKMODE lockmode,
! 					List **child_oids, List **child_inhs)
  {
! 	List	   *rel_oids, *rel_inhs;
  	ListCell   *l;
  
  	/*
***************
*** 167,193 **** find_all_inheritors(Oid parentrelId, LOCKMODE lockmode)
  	 * children.  This is a bit tricky but works because the foreach() macro
  	 * doesn't fetch the next list element until the bottom of the loop.
  	 */
! 	rels_list = list_make1_oid(parentrelId);
  
! 	foreach(l, rels_list)
  	{
! 		Oid			currentrel = lfirst_oid(l);
! 		List	   *currentchildren;
  
  		/* Get the direct children of this rel */
! 		currentchildren = find_inheritance_children(currentrel, lockmode);
  
! 		/*
! 		 * Add to the queue only those children not already seen. This avoids
! 		 * making duplicate entries in case of multiple inheritance paths from
! 		 * the same parent.  (It'll also keep us from getting into an infinite
! 		 * loop, though theoretically there can't be any cycles in the
! 		 * inheritance graph anyway.)
! 		 */
! 		rels_list = list_concat_unique_oid(rels_list, currentchildren);
  	}
  
! 	return rels_list;
  }
  
  
--- 198,257 ----
  	 * children.  This is a bit tricky but works because the foreach() macro
  	 * doesn't fetch the next list element until the bottom of the loop.
  	 */
! 	rel_oids = list_make1_oid(parentrelId);
! 	rel_inhs = list_make1_int(0);
  
! 	foreach(l, rel_oids)
  	{
! 		Oid			cur_relid = lfirst_oid(l);
! 		Oid			temp_relid;
! 		List	   *temp;
! 		ListCell   *lt, *lo, *li;
  
  		/* Get the direct children of this rel */
! 		temp = find_inheritance_children(cur_relid, lockmode);
  
! 		foreach(lt, temp)
! 		{
! 			Oid		temp_relid = lfirst_oid(lt);
! 			bool	found = false;
! 
! 			/*
! 			 * Add to the queue only those children not already seen.
! 			 * This avoids making duplicate entries in case of multiple
! 			 * inheritance paths from the same parent.  (It'll also keep
! 			 * us from getting into an infinite loop, though theoretically
! 			 * there can't be any cycles in the inheritance graph anyway.)
! 			 */
! 			forboth(lo, rel_oids, li, rel_inhs)
! 			{
! 				/*
! 				 * Increment its expected inhcount, if the child relation
! 				 * is already chained to the rel_oids due to the diamond
! 				 * inheritance tree. In this case, we don't need to chain
! 				 * a new entry into the list.
! 				 */
! 
! 				if (lfirst_oid(lo) == temp_relid)
! 				{
! 					lfirst_int(li)++;
! 					found = true;
! 					break;
! 				}
! 			}
! 
! 			if (!found)
! 			{
! 				rel_oids = lappend_oid(rel_oids, temp_relid);
! 				rel_inhs = lappend_int(rel_inhs, 1);
! 			}
! 		}
  	}
  
! 	if (child_oids)
! 		*child_oids = rel_oids;
! 	if (child_inhs)
! 		*child_inhs = rel_inhs;
  }
  
  
*** a/src/backend/commands/alter.c
--- b/src/backend/commands/alter.c
***************
*** 125,130 **** ExecRenameStmt(RenameStmt *stmt)
--- 125,131 ----
  						renameatt(relid,
  								  stmt->subname,		/* old att name */
  								  stmt->newname,		/* new att name */
+ 								  0,			/* expected inhcount */
  								  interpretInhOption(stmt->relation->inhOpt),	/* recursive? */
  								  false);		/* recursing already? */
  						break;
*** a/src/backend/commands/analyze.c
--- b/src/backend/commands/analyze.c
***************
*** 1390,1396 **** acquire_inherited_sample_rows(Relation onerel, HeapTuple *rows, int targrows,
  	 * Find all members of inheritance set.  We only need AccessShareLock on
  	 * the children.
  	 */
! 	tableOIDs = find_all_inheritors(RelationGetRelid(onerel), AccessShareLock);
  
  	/*
  	 * Check that there's at least one descendant, else fail.  This could
--- 1390,1397 ----
  	 * Find all members of inheritance set.  We only need AccessShareLock on
  	 * the children.
  	 */
! 	find_all_inheritors(RelationGetRelid(onerel), AccessShareLock,
! 						&tableOIDs, NULL);
  
  	/*
  	 * Check that there's at least one descendant, else fail.  This could
*** a/src/backend/commands/tablecmds.c
--- b/src/backend/commands/tablecmds.c
***************
*** 824,830 **** ExecuteTruncate(TruncateStmt *stmt)
  			ListCell   *child;
  			List	   *children;
  
! 			children = find_all_inheritors(myrelid, AccessExclusiveLock);
  
  			foreach(child, children)
  			{
--- 824,831 ----
  			ListCell   *child;
  			List	   *children;
  
! 			find_all_inheritors(myrelid, AccessExclusiveLock,
! 								&children, NULL);
  
  			foreach(child, children)
  			{
***************
*** 1942,1947 **** void
--- 1943,1949 ----
  renameatt(Oid myrelid,
  		  const char *oldattname,
  		  const char *newattname,
+ 		  int expected_inhcount,
  		  bool recurse,
  		  bool recursing)
  {
***************
*** 1987,2010 **** renameatt(Oid myrelid,
  	 */
  	if (recurse)
  	{
! 		ListCell   *child;
! 		List	   *children;
! 
! 		children = find_all_inheritors(myrelid, AccessExclusiveLock);
  
  		/*
  		 * find_all_inheritors does the recursive search of the inheritance
  		 * hierarchy, so all we have to do is process all of the relids in the
  		 * list that it returns.
  		 */
! 		foreach(child, children)
  		{
! 			Oid			childrelid = lfirst_oid(child);
  
! 			if (childrelid == myrelid)
  				continue;
  			/* note we need not recurse again */
! 			renameatt(childrelid, oldattname, newattname, false, true);
  		}
  	}
  	else
--- 1989,2019 ----
  	 */
  	if (recurse)
  	{
! 		List	   *child_oids, *child_inhs;
! 		ListCell   *lo, *li;
  
  		/*
  		 * find_all_inheritors does the recursive search of the inheritance
  		 * hierarchy, so all we have to do is process all of the relids in the
  		 * list that it returns.
+ 		 * Also, we need the list of expected inhcount, since it has to be
+ 		 * forbidden to rename a column that is inherited from different
+ 		 * origin. (Note that we allow diamond-inheritance tree)
  		 */
! 		find_all_inheritors(myrelid, AccessExclusiveLock,
! 							&child_oids, &child_inhs);
! 
! 		forboth(lo, child_oids, li, child_inhs)
  		{
! 			Oid		child_relid = lfirst_oid(lo);
! 			int		child_inhcount = lfirst_int(li);
  
! 			if (child_relid == myrelid)
  				continue;
+ 
  			/* note we need not recurse again */
! 			renameatt(child_relid, oldattname, newattname,
! 					  child_inhcount, false, true);
  		}
  	}
  	else
***************
*** 2048,2053 **** renameatt(Oid myrelid,
--- 2057,2073 ----
  				 errmsg("cannot rename inherited column \"%s\"",
  						oldattname)));
  
+ 	/*
+ 	 * if the attribute is multiple inherited, forbid the renaming,
+ 	 * even if we are already inside a recursive rename, because we
+ 	 * have no reasonable way to keep its integrity.
+ 	 */
+ 	if (attform->attinhcount > expected_inhcount)
+ 		ereport(ERROR,
+ 				(errcode(ERRCODE_INVALID_TABLE_DEFINITION),
+ 				 (errmsg("cannot rename multiple inherited column \"%s\"",
+ 						 oldattname))));
+ 
  	/* new name should not already exist */
  
  	/* this test is deliberately not attisdropped-aware */
***************
*** 3410,3416 **** ATSimpleRecursion(List **wqueue, Relation rel,
  		ListCell   *child;
  		List	   *children;
  
! 		children = find_all_inheritors(relid, AccessExclusiveLock);
  
  		/*
  		 * find_all_inheritors does the recursive search of the inheritance
--- 3430,3436 ----
  		ListCell   *child;
  		List	   *children;
  
! 		find_all_inheritors(relid, AccessExclusiveLock, &children, NULL);
  
  		/*
  		 * find_all_inheritors does the recursive search of the inheritance
***************
*** 7232,7239 **** ATExecAddInherit(Relation child_rel, RangeVar *parent)
  	 *
  	 * We use weakest lock we can on child's children, namely AccessShareLock.
  	 */
! 	children = find_all_inheritors(RelationGetRelid(child_rel),
! 								   AccessShareLock);
  
  	if (list_member_oid(children, RelationGetRelid(parent_rel)))
  		ereport(ERROR,
--- 7252,7259 ----
  	 *
  	 * We use weakest lock we can on child's children, namely AccessShareLock.
  	 */
! 	find_all_inheritors(RelationGetRelid(child_rel), AccessShareLock,
! 						&children, NULL);
  
  	if (list_member_oid(children, RelationGetRelid(parent_rel)))
  		ereport(ERROR,
*** a/src/backend/optimizer/prep/prepunion.c
--- b/src/backend/optimizer/prep/prepunion.c
***************
*** 1180,1186 **** expand_inherited_rtentry(PlannerInfo *root, RangeTblEntry *rte, Index rti)
  		lockmode = AccessShareLock;
  
  	/* Scan for all members of inheritance set, acquire needed locks */
! 	inhOIDs = find_all_inheritors(parentOID, lockmode);
  
  	/*
  	 * Check that there's at least one descendant, else treat as no-child
--- 1180,1186 ----
  		lockmode = AccessShareLock;
  
  	/* Scan for all members of inheritance set, acquire needed locks */
! 	find_all_inheritors(parentOID, lockmode, &inhOIDs, NULL);
  
  	/*
  	 * Check that there's at least one descendant, else treat as no-child
*** a/src/include/catalog/pg_inherits_fn.h
--- b/src/include/catalog/pg_inherits_fn.h
***************
*** 18,24 ****
  #include "storage/lock.h"
  
  extern List *find_inheritance_children(Oid parentrelId, LOCKMODE lockmode);
! extern List *find_all_inheritors(Oid parentrelId, LOCKMODE lockmode);
  extern bool has_subclass(Oid relationId);
  extern bool typeInheritsFrom(Oid subclassTypeId, Oid superclassTypeId);
  
--- 18,25 ----
  #include "storage/lock.h"
  
  extern List *find_inheritance_children(Oid parentrelId, LOCKMODE lockmode);
! extern void find_all_inheritors(Oid parentrelId, LOCKMODE lockmode,
! 								List **child_oids, List **child_inhs);
  extern bool has_subclass(Oid relationId);
  extern bool typeInheritsFrom(Oid subclassTypeId, Oid superclassTypeId);
  
*** a/src/include/commands/tablecmds.h
--- b/src/include/commands/tablecmds.h
***************
*** 42,47 **** extern void ExecuteTruncate(TruncateStmt *stmt);
--- 42,48 ----
  extern void renameatt(Oid myrelid,
  		  const char *oldattname,
  		  const char *newattname,
+ 		  int expected_inhcount,
  		  bool recurse,
  		  bool recursing);
  
*** a/src/test/regress/expected/inherit.out
--- b/src/test/regress/expected/inherit.out
***************
*** 1053,1055 **** NOTICE:  merging column "a" with inherited definition
--- 1053,1148 ----
  ERROR:  column "a" has a storage parameter conflict
  DETAIL:  MAIN versus EXTENDED
  DROP TABLE t1, t2, t3, t4, t12_storage, t12_comments, t1_inh, t13_inh, t13_like, t_all;
+ -- Test for renaming in simple multiple inheritance
+ CREATE TABLE t1 (a int, b int);
+ CREATE TABLE s1 (b int, c int);
+ CREATE TABLE ts (d int) INHERITS (t1, s1);
+ NOTICE:  merging multiple inherited definitions of column "b"
+ ALTER TABLE t1 RENAME a TO aa;
+ ALTER TABLE t1 RENAME b TO bb;                -- to be failed
+ ERROR:  cannot rename multiple inherited column "b"
+ ALTER TABLE ts RENAME aa TO aaa;      -- to be failed
+ ERROR:  cannot rename inherited column "aa"
+ ALTER TABLE ts RENAME d TO dd;
+ \d+ ts
+                   Table "public.ts"
+  Column |  Type   | Modifiers | Storage | Description 
+ --------+---------+-----------+---------+-------------
+  aa     | integer |           | plain   | 
+  b      | integer |           | plain   | 
+  c      | integer |           | plain   | 
+  dd     | integer |           | plain   | 
+ Inherits: t1,
+           s1
+ Has OIDs: no
+ 
+ DROP TABLE ts;
+ -- Test for renaming in diamond inheritance
+ CREATE TABLE t2 (x int) INHERITS (t1);
+ CREATE TABLE t3 (y int) INHERITS (t1);
+ CREATE TABLE t4 (z int) INHERITS (t2, t3);
+ NOTICE:  merging multiple inherited definitions of column "aa"
+ NOTICE:  merging multiple inherited definitions of column "b"
+ ALTER TABLE t1 RENAME aa TO aaa;
+ \d+ t4
+                   Table "public.t4"
+  Column |  Type   | Modifiers | Storage | Description 
+ --------+---------+-----------+---------+-------------
+  aaa    | integer |           | plain   | 
+  b      | integer |           | plain   | 
+  x      | integer |           | plain   | 
+  y      | integer |           | plain   | 
+  z      | integer |           | plain   | 
+ Inherits: t2,
+           t3
+ Has OIDs: no
+ 
+ CREATE TABLE ts (d int) INHERITS (t2, s1);
+ NOTICE:  merging multiple inherited definitions of column "b"
+ ALTER TABLE t1 RENAME aaa TO aaaa;
+ ALTER TABLE t1 RENAME b TO bb;                -- to be failed
+ ERROR:  cannot rename multiple inherited column "b"
+ \d+ ts
+                   Table "public.ts"
+  Column |  Type   | Modifiers | Storage | Description 
+ --------+---------+-----------+---------+-------------
+  aaaa   | integer |           | plain   | 
+  b      | integer |           | plain   | 
+  x      | integer |           | plain   | 
+  c      | integer |           | plain   | 
+  d      | integer |           | plain   | 
+ Inherits: t2,
+           s1
+ Has OIDs: no
+ 
+ WITH RECURSIVE r AS (
+   SELECT 't1'::regclass AS inhrelid
+ UNION ALL
+   SELECT c.inhrelid FROM pg_inherits c, r WHERE r.inhrelid = c.inhparent
+ )
+ SELECT attrelid::regclass, attname, attinhcount, r.expected
+   FROM (SELECT inhrelid, count(*) AS expected FROM pg_inherits
+         WHERE inhparent IN (SELECT inhrelid FROM r) GROUP BY inhrelid) r
+   JOIN pg_attribute a ON r.inhrelid = a.attrelid WHERE attislocal = false;
+  attrelid | attname | attinhcount | expected 
+ ----------+---------+-------------+----------
+  t2       | b       |           1 |        1
+  t3       | b       |           1 |        1
+  t4       | b       |           2 |        2
+  t4       | x       |           1 |        2
+  t4       | y       |           1 |        2
+  ts       | b       |           2 |        1
+  ts       | x       |           1 |        1
+  ts       | c       |           1 |        1
+  t2       | aaaa    |           1 |        1
+  t3       | aaaa    |           1 |        1
+  t4       | aaaa    |           2 |        2
+  ts       | aaaa    |           1 |        1
+ (12 rows)
+ 
+ DROP TABLE t1, s1 CASCADE;
+ NOTICE:  drop cascades to 4 other objects
+ DETAIL:  drop cascades to table t2
+ drop cascades to table ts
+ drop cascades to table t3
+ drop cascades to table t4
*** a/src/test/regress/sql/inherit.sql
--- b/src/test/regress/sql/inherit.sql
***************
*** 334,336 **** CREATE TABLE inh_error1 () INHERITS (t1, t4);
--- 334,374 ----
  CREATE TABLE inh_error2 (LIKE t4 INCLUDING STORAGE) INHERITS (t1);
  
  DROP TABLE t1, t2, t3, t4, t12_storage, t12_comments, t1_inh, t13_inh, t13_like, t_all;
+ 
+ -- Test for renaming in simple multiple inheritance
+ CREATE TABLE t1 (a int, b int);
+ CREATE TABLE s1 (b int, c int);
+ CREATE TABLE ts (d int) INHERITS (t1, s1);
+ 
+ ALTER TABLE t1 RENAME a TO aa;
+ ALTER TABLE t1 RENAME b TO bb;                -- to be failed
+ ALTER TABLE ts RENAME aa TO aaa;      -- to be failed
+ ALTER TABLE ts RENAME d TO dd;
+ \d+ ts
+ 
+ DROP TABLE ts;
+ 
+ -- Test for renaming in diamond inheritance
+ CREATE TABLE t2 (x int) INHERITS (t1);
+ CREATE TABLE t3 (y int) INHERITS (t1);
+ CREATE TABLE t4 (z int) INHERITS (t2, t3);
+ 
+ ALTER TABLE t1 RENAME aa TO aaa;
+ \d+ t4
+ 
+ CREATE TABLE ts (d int) INHERITS (t2, s1);
+ ALTER TABLE t1 RENAME aaa TO aaaa;
+ ALTER TABLE t1 RENAME b TO bb;                -- to be failed
+ \d+ ts
+ 
+ WITH RECURSIVE r AS (
+   SELECT 't1'::regclass AS inhrelid
+ UNION ALL
+   SELECT c.inhrelid FROM pg_inherits c, r WHERE r.inhrelid = c.inhparent
+ )
+ SELECT attrelid::regclass, attname, attinhcount, r.expected
+   FROM (SELECT inhrelid, count(*) AS expected FROM pg_inherits
+         WHERE inhparent IN (SELECT inhrelid FROM r) GROUP BY inhrelid) r
+   JOIN pg_attribute a ON r.inhrelid = a.attrelid WHERE attislocal = false;
+ 
+ DROP TABLE t1, s1 CASCADE;