v1-0001-fix-table-privilege-checks.patch

diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index d1f5dcd8be..95b0f8c7b0 100644
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -13,6 +13,27 @@
  * NOTES
  *	  See acl.h.
  *
+ *	  The xxx_aclmask() functions in this file are wrappers around
+ *	  acl.c's aclmask() function; see that for basic usage information.
+ *	  The wrapper functions add object-type-specific lookup capability.
+ *	  Generally, they will throw error if the object doesn't exist.
+ *
+ *	  The xxx_aclmask_ext() functions add the ability to not throw
+ *	  error if the object doesn't exist.  If their "is_missing" argument
+ *	  isn't NULL, then when the object isn't found they will set
+ *	  *is_missing = true and return zero (no privileges) instead of
+ *	  throwing an error.  Caller must initialize *is_missing = false.
+ *
+ *	  The xxx_aclcheck() functions are simplified wrappers around the
+ *	  corresponding xxx_aclmask() functions, simply returning ACLCHECK_OK
+ *	  if any of the privileges specified in "mode" are held, and otherwise
+ *	  a suitable error code (in practice, always ACLCHECK_NO_PRIV).
+ *	  Again, they will throw error if the object doesn't exist.
+ *
+ *	  The xxx_aclcheck_ext() functions add the ability to not throw
+ *	  error if the object doesn't exist.  Their "is_missing" argument
+ *	  works similarly to the xxx_aclmask_ext() functions.
+ *
  *-------------------------------------------------------------------------
  */
 #include "postgres.h"
@@ -3149,10 +3170,7 @@ pg_attribute_aclmask(Oid table_oid, AttrNumber attnum, Oid roleid,
 }
 
 /*
- * Routine for examining a user's privileges for a column
- *
- * Does the bulk of the work for pg_attribute_aclmask(), and allows other
- * callers to avoid the missing attribute ERROR when is_missing is non-NULL.
+ * Routine for examining a user's privileges for a column, with is_missing
  */
 static AclMode
 pg_attribute_aclmask_ext(Oid table_oid, AttrNumber attnum, Oid roleid,
@@ -3226,15 +3244,24 @@ pg_attribute_aclmask_ext(Oid table_oid, AttrNumber attnum, Oid roleid,
 	 * Must get the relation's ownerId from pg_class.  Since we already found
 	 * a pg_attribute entry, the only likely reason for this to fail is that a
 	 * concurrent DROP of the relation committed since then (which could only
-	 * happen if we don't have lock on the relation).  We prefer to report "no
-	 * privileges" rather than failing in such a case, so as to avoid unwanted
-	 * failures in has_column_privilege() tests.
+	 * happen if we don't have lock on the relation).  Treat that similarly to
+	 * not finding the attribute entry.
 	 */
 	classTuple = SearchSysCache1(RELOID, ObjectIdGetDatum(table_oid));
 	if (!HeapTupleIsValid(classTuple))
 	{
 		ReleaseSysCache(attTuple);
-		return 0;
+		if (is_missing != NULL)
+		{
+			/* return "no privileges" instead of throwing an error */
+			*is_missing = true;
+			return 0;
+		}
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_TABLE),
+					 errmsg("relation with OID %u does not exist",
+							table_oid)));
 	}
 	classForm = (Form_pg_class) GETSTRUCT(classTuple);
 
@@ -3267,10 +3294,7 @@ pg_class_aclmask(Oid table_oid, Oid roleid,
 }
 
 /*
- * Routine for examining a user's privileges for a table
- *
- * Does the bulk of the work for pg_class_aclmask(), and allows other
- * callers to avoid the missing relation ERROR when is_missing is non-NULL.
+ * Routine for examining a user's privileges for a table, with is_missing
  */
 static AclMode
 pg_class_aclmask_ext(Oid table_oid, Oid roleid, AclMode mask,
@@ -3784,10 +3808,8 @@ pg_attribute_aclcheck(Oid table_oid, AttrNumber attnum,
 
 
 /*
- * Exported routine for checking a user's access privileges to a column
- *
- * Does the bulk of the work for pg_attribute_aclcheck(), and allows other
- * callers to avoid the missing attribute ERROR when is_missing is non-NULL.
+ * Exported routine for checking a user's access privileges to a column,
+ * with is_missing
  */
 AclResult
 pg_attribute_aclcheck_ext(Oid table_oid, AttrNumber attnum,
@@ -3822,23 +3844,47 @@ pg_attribute_aclcheck_ext(Oid table_oid, AttrNumber attnum,
 AclResult
 pg_attribute_aclcheck_all(Oid table_oid, Oid roleid, AclMode mode,
 						  AclMaskHow how)
+{
+	return pg_attribute_aclcheck_all_ext(table_oid, roleid, mode, how, NULL);
+}
+
+/*
+ * Exported routine for checking a user's access privileges to any/all columns,
+ * with is_missing
+ */
+AclResult
+pg_attribute_aclcheck_all_ext(Oid table_oid, Oid roleid,
+							  AclMode mode, AclMaskHow how,
+							  bool *is_missing)
 {
 	AclResult	result;
 	HeapTuple	classTuple;
 	Form_pg_class classForm;
+	Oid			ownerId;
 	AttrNumber	nattrs;
 	AttrNumber	curr_att;
 
 	/*
-	 * Must fetch pg_class row to check number of attributes.  As in
-	 * pg_attribute_aclmask, we prefer to return "no privileges" instead of
-	 * throwing an error if we get any unexpected lookup errors.
+	 * Must fetch pg_class row to get owner ID and number of attributes.
 	 */
 	classTuple = SearchSysCache1(RELOID, ObjectIdGetDatum(table_oid));
 	if (!HeapTupleIsValid(classTuple))
-		return ACLCHECK_NO_PRIV;
+	{
+		if (is_missing != NULL)
+		{
+			/* return "no privileges" instead of throwing an error */
+			*is_missing = true;
+			return ACLCHECK_NO_PRIV;
+		}
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_TABLE),
+					 errmsg("relation with OID %u does not exist",
+							table_oid)));
+	}
 	classForm = (Form_pg_class) GETSTRUCT(classTuple);
 
+	ownerId = classForm->relowner;
 	nattrs = classForm->relnatts;
 
 	ReleaseSysCache(classTuple);
@@ -3852,11 +3898,20 @@ pg_attribute_aclcheck_all(Oid table_oid, Oid roleid, AclMode mode,
 	for (curr_att = 1; curr_att <= nattrs; curr_att++)
 	{
 		HeapTuple	attTuple;
+		Datum		aclDatum;
+		bool		isNull;
+		Acl		   *acl;
 		AclMode		attmask;
 
 		attTuple = SearchSysCache2(ATTNUM,
 								   ObjectIdGetDatum(table_oid),
 								   Int16GetDatum(curr_att));
+
+		/*
+		 * Lookup failure probably indicates that the table was just dropped,
+		 * but we'll treat it the same as a dropped column rather than
+		 * throwing error.
+		 */
 		if (!HeapTupleIsValid(attTuple))
 			continue;
 
@@ -3867,16 +3922,27 @@ pg_attribute_aclcheck_all(Oid table_oid, Oid roleid, AclMode mode,
 			continue;
 		}
 
+		aclDatum = SysCacheGetAttr(ATTNUM, attTuple, Anum_pg_attribute_attacl,
+								   &isNull);
+
 		/*
 		 * Here we hard-wire knowledge that the default ACL for a column
 		 * grants no privileges, so that we can fall out quickly in the very
 		 * common case where attacl is null.
 		 */
-		if (heap_attisnull(attTuple, Anum_pg_attribute_attacl, NULL))
+		if (isNull)
 			attmask = 0;
 		else
-			attmask = pg_attribute_aclmask(table_oid, curr_att, roleid,
-										   mode, ACLMASK_ANY);
+		{
+			/* detoast column's ACL if necessary */
+			acl = DatumGetAclP(aclDatum);
+
+			attmask = aclmask(acl, roleid, ownerId, mode, ACLMASK_ANY);
+
+			/* if we have a detoasted copy, free it */
+			if ((Pointer) acl != DatumGetPointer(aclDatum))
+				pfree(acl);
+		}
 
 		ReleaseSysCache(attTuple);
 
@@ -3911,10 +3977,8 @@ pg_class_aclcheck(Oid table_oid, Oid roleid, AclMode mode)
 }
 
 /*
- * Exported routine for checking a user's access privileges to a table
- *
- * Does the bulk of the work for pg_class_aclcheck(), and allows other
- * callers to avoid the missing relation ERROR when is_missing is non-NULL.
+ * Exported routine for checking a user's access privileges to a table,
+ * with is_missing
  */
 AclResult
 pg_class_aclcheck_ext(Oid table_oid, Oid roleid,
diff --git a/src/backend/utils/adt/acl.c b/src/backend/utils/adt/acl.c
index 27eabb80ab..809cb6f03f 100644
--- a/src/backend/utils/adt/acl.c
+++ b/src/backend/utils/adt/acl.c
@@ -1915,14 +1915,15 @@ has_table_privilege_name_id(PG_FUNCTION_ARGS)
 	Oid			roleid;
 	AclMode		mode;
 	AclResult	aclresult;
+	bool		is_missing = false;
 
 	roleid = get_role_oid_or_public(NameStr(*username));
 	mode = convert_table_priv_string(priv_type_text);
 
-	if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(tableoid)))
-		PG_RETURN_NULL();
+	aclresult = pg_class_aclcheck_ext(tableoid, roleid, mode, &is_missing);
 
-	aclresult = pg_class_aclcheck(tableoid, roleid, mode);
+	if (is_missing)
+		PG_RETURN_NULL();
 
 	PG_RETURN_BOOL(aclresult == ACLCHECK_OK);
 }
@@ -1941,14 +1942,15 @@ has_table_privilege_id(PG_FUNCTION_ARGS)
 	Oid			roleid;
 	AclMode		mode;
 	AclResult	aclresult;
+	bool		is_missing = false;
 
 	roleid = GetUserId();
 	mode = convert_table_priv_string(priv_type_text);
 
-	if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(tableoid)))
-		PG_RETURN_NULL();
+	aclresult = pg_class_aclcheck_ext(tableoid, roleid, mode, &is_missing);
 
-	aclresult = pg_class_aclcheck(tableoid, roleid, mode);
+	if (is_missing)
+		PG_RETURN_NULL();
 
 	PG_RETURN_BOOL(aclresult == ACLCHECK_OK);
 }
@@ -1989,13 +1991,14 @@ has_table_privilege_id_id(PG_FUNCTION_ARGS)
 	text	   *priv_type_text = PG_GETARG_TEXT_PP(2);
 	AclMode		mode;
 	AclResult	aclresult;
+	bool		is_missing = false;
 
 	mode = convert_table_priv_string(priv_type_text);
 
-	if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(tableoid)))
-		PG_RETURN_NULL();
+	aclresult = pg_class_aclcheck_ext(tableoid, roleid, mode, &is_missing);
 
-	aclresult = pg_class_aclcheck(tableoid, roleid, mode);
+	if (is_missing)
+		PG_RETURN_NULL();
 
 	PG_RETURN_BOOL(aclresult == ACLCHECK_OK);
 }
@@ -2134,6 +2137,7 @@ has_sequence_privilege_name_id(PG_FUNCTION_ARGS)
 	AclMode		mode;
 	AclResult	aclresult;
 	char		relkind;
+	bool		is_missing = false;
 
 	roleid = get_role_oid_or_public(NameStr(*username));
 	mode = convert_sequence_priv_string(priv_type_text);
@@ -2146,7 +2150,10 @@ has_sequence_privilege_name_id(PG_FUNCTION_ARGS)
 				 errmsg("\"%s\" is not a sequence",
 						get_rel_name(sequenceoid))));
 
-	aclresult = pg_class_aclcheck(sequenceoid, roleid, mode);
+	aclresult = pg_class_aclcheck_ext(sequenceoid, roleid, mode, &is_missing);
+
+	if (is_missing)
+		PG_RETURN_NULL();
 
 	PG_RETURN_BOOL(aclresult == ACLCHECK_OK);
 }
@@ -2166,6 +2173,7 @@ has_sequence_privilege_id(PG_FUNCTION_ARGS)
 	AclMode		mode;
 	AclResult	aclresult;
 	char		relkind;
+	bool		is_missing = false;
 
 	roleid = GetUserId();
 	mode = convert_sequence_priv_string(priv_type_text);
@@ -2178,7 +2186,11 @@ has_sequence_privilege_id(PG_FUNCTION_ARGS)
 				 errmsg("\"%s\" is not a sequence",
 						get_rel_name(sequenceoid))));
 
-	aclresult = pg_class_aclcheck(sequenceoid, roleid, mode);
+
+	aclresult = pg_class_aclcheck_ext(sequenceoid, roleid, mode, &is_missing);
+
+	if (is_missing)
+		PG_RETURN_NULL();
 
 	PG_RETURN_BOOL(aclresult == ACLCHECK_OK);
 }
@@ -2225,6 +2237,7 @@ has_sequence_privilege_id_id(PG_FUNCTION_ARGS)
 	AclMode		mode;
 	AclResult	aclresult;
 	char		relkind;
+	bool		is_missing = false;
 
 	mode = convert_sequence_priv_string(priv_type_text);
 	relkind = get_rel_relkind(sequenceoid);
@@ -2236,7 +2249,11 @@ has_sequence_privilege_id_id(PG_FUNCTION_ARGS)
 				 errmsg("\"%s\" is not a sequence",
 						get_rel_name(sequenceoid))));
 
-	aclresult = pg_class_aclcheck(sequenceoid, roleid, mode);
+
+	aclresult = pg_class_aclcheck_ext(sequenceoid, roleid, mode, &is_missing);
+
+	if (is_missing)
+		PG_RETURN_NULL();
 
 	PG_RETURN_BOOL(aclresult == ACLCHECK_OK);
 }
@@ -2345,18 +2362,22 @@ has_any_column_privilege_name_id(PG_FUNCTION_ARGS)
 	Oid			roleid;
 	AclMode		mode;
 	AclResult	aclresult;
+	bool		is_missing = false;
 
 	roleid = get_role_oid_or_public(NameStr(*username));
 	mode = convert_column_priv_string(priv_type_text);
 
-	if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(tableoid)))
-		PG_RETURN_NULL();
-
 	/* First check at table level, then examine each column if needed */
-	aclresult = pg_class_aclcheck(tableoid, roleid, mode);
+	aclresult = pg_class_aclcheck_ext(tableoid, roleid, mode, &is_missing);
 	if (aclresult != ACLCHECK_OK)
-		aclresult = pg_attribute_aclcheck_all(tableoid, roleid, mode,
-											  ACLMASK_ANY);
+	{
+		if (is_missing)
+			PG_RETURN_NULL();
+		aclresult = pg_attribute_aclcheck_all_ext(tableoid, roleid, mode,
+												  ACLMASK_ANY, &is_missing);
+		if (is_missing)
+			PG_RETURN_NULL();
+	}
 
 	PG_RETURN_BOOL(aclresult == ACLCHECK_OK);
 }
@@ -2375,18 +2396,22 @@ has_any_column_privilege_id(PG_FUNCTION_ARGS)
 	Oid			roleid;
 	AclMode		mode;
 	AclResult	aclresult;
+	bool		is_missing = false;
 
 	roleid = GetUserId();
 	mode = convert_column_priv_string(priv_type_text);
 
-	if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(tableoid)))
-		PG_RETURN_NULL();
-
 	/* First check at table level, then examine each column if needed */
-	aclresult = pg_class_aclcheck(tableoid, roleid, mode);
+	aclresult = pg_class_aclcheck_ext(tableoid, roleid, mode, &is_missing);
 	if (aclresult != ACLCHECK_OK)
-		aclresult = pg_attribute_aclcheck_all(tableoid, roleid, mode,
-											  ACLMASK_ANY);
+	{
+		if (is_missing)
+			PG_RETURN_NULL();
+		aclresult = pg_attribute_aclcheck_all_ext(tableoid, roleid, mode,
+												  ACLMASK_ANY, &is_missing);
+		if (is_missing)
+			PG_RETURN_NULL();
+	}
 
 	PG_RETURN_BOOL(aclresult == ACLCHECK_OK);
 }
@@ -2431,17 +2456,21 @@ has_any_column_privilege_id_id(PG_FUNCTION_ARGS)
 	text	   *priv_type_text = PG_GETARG_TEXT_PP(2);
 	AclMode		mode;
 	AclResult	aclresult;
+	bool		is_missing = false;
 
 	mode = convert_column_priv_string(priv_type_text);
 
-	if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(tableoid)))
-		PG_RETURN_NULL();
-
 	/* First check at table level, then examine each column if needed */
-	aclresult = pg_class_aclcheck(tableoid, roleid, mode);
+	aclresult = pg_class_aclcheck_ext(tableoid, roleid, mode, &is_missing);
 	if (aclresult != ACLCHECK_OK)
-		aclresult = pg_attribute_aclcheck_all(tableoid, roleid, mode,
-											  ACLMASK_ANY);
+	{
+		if (is_missing)
+			PG_RETURN_NULL();
+		aclresult = pg_attribute_aclcheck_all_ext(tableoid, roleid, mode,
+												  ACLMASK_ANY, &is_missing);
+		if (is_missing)
+			PG_RETURN_NULL();
+	}
 
 	PG_RETURN_BOOL(aclresult == ACLCHECK_OK);
 }
diff --git a/src/include/utils/acl.h b/src/include/utils/acl.h
index aba1afa971..331a87d0e6 100644
--- a/src/include/utils/acl.h
+++ b/src/include/utils/acl.h
@@ -250,6 +250,9 @@ extern AclResult pg_attribute_aclcheck_ext(Oid table_oid, AttrNumber attnum,
 										   bool *is_missing);
 extern AclResult pg_attribute_aclcheck_all(Oid table_oid, Oid roleid,
 										   AclMode mode, AclMaskHow how);
+extern AclResult pg_attribute_aclcheck_all_ext(Oid table_oid, Oid roleid,
+											   AclMode mode, AclMaskHow how,
+											   bool *is_missing);
 extern AclResult pg_class_aclcheck(Oid table_oid, Oid roleid, AclMode mode);
 extern AclResult pg_class_aclcheck_ext(Oid table_oid, Oid roleid,
 									   AclMode mode, bool *is_missing);