Add finer control in EnsureTablePermission

Previously this function was only checking ACL at table level, however
for propagating GRANT on attributes, it is required to lock the shard
while the user may not have table privilege at all (but attribute level
ones).

In order to solve that, I add a new parameter to the function: the ACL
mask to be used when checking attributes acl:

-----
Check that the current user has `mode` permissions on relationId.

If not, also check relationId's attributes with `mask`, error out
privileges are not defined.

ACL mask is used because we assume that user has enought privilege
to distribute a table when either ACL_INSERT on the TABLE or
ACL_INSERT on ALL attributes.

In other situations, having a single attribute privilege is enough.
-----

src/backend/distributed/metadata/metadata_utility.c

@@ -2428,12 +2428,17 @@ UpdateNoneDistTableMetadata(Oid relationId, char replicationModel, uint32 coloca
 
 
 /*
- * Check that the current user has `mode` permissions on relationId or on at
- * least one relationId's attribute, error out if not.
+ * Check that the current user has `mode` permissions on relationId.
+ * If not, also check relationId's attributes with `mask`, error out
+ * privileges are not defined.
+ * ACL mask is used because we assume that user has enought privilege
+ * to distribute a table when either ACL_INSERT on the TABLE or
+ * ACL_INSERT on ALL attributes.
+ * In other situations, having a single attribute privilege is enough.
  * Superusers always have such permissions.
  */
 void
-EnsureTablePermissions(Oid relationId, AclMode mode)
+EnsureTablePermissions(Oid relationId, AclMode mode, AclMaskHow mask)
 {
 	AclResult aclresult = pg_class_aclcheck(relationId, GetUserId(), mode);
 

src/backend/distributed/metadata/node_metadata.c

@@ -1522,7 +1522,7 @@ get_shard_id_for_distribution_column(PG_FUNCTION_ARGS)
 	}
 
 	Oid relationId = PG_GETARG_OID(0);
-	EnsureTablePermissions(relationId, ACL_SELECT);
+	EnsureTablePermissions(relationId, ACL_SELECT, ACLMASK_ANY);
 
 	if (!IsCitusTable(relationId))
 	{

src/backend/distributed/operations/stage_protocol.c

@@ -108,7 +108,7 @@ master_create_empty_shard(PG_FUNCTION_ARGS)
 
 	Oid relationId = ResolveRelationId(relationNameText, false);
 
-	EnsureTablePermissions(relationId, ACL_INSERT);
+	EnsureTablePermissions(relationId, ACL_INSERT, ACLMASK_ALL);
 	CheckDistributedTable(relationId);
 
 	/*

src/backend/distributed/utils/resource_lock.c

@@ -222,10 +222,12 @@ lock_shard_resources(PG_FUNCTION_ARGS)
 	 * on the executor. However, for INSERTs, the user might have only
 	 * INSERTs granted, so add a special case for it.
 	 */
-	AclMode aclMask = ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE;
+	AclMode aclMode = ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE;
+	AclMaskHow aclMaskHow = ACLMASK_ANY;
+
 	if (lockMode == RowExclusiveLock)
 	{
-		aclMask |= ACL_INSERT;
+		aclMode |= ACL_INSERT;
 	}
 
 	for (int shardIdIndex = 0; shardIdIndex < shardIdCount; shardIdIndex++)
@@ -254,7 +256,7 @@ lock_shard_resources(PG_FUNCTION_ARGS)
 
 		if (!SkipAdvisoryLockPermissionChecks)
 		{
-			EnsureTablePermissions(relationId, aclMask);
+			EnsureTablePermissions(relationId, aclMode, aclMaskHow);
 		}
 
 		LockShardResource(shardId, lockMode);

src/include/distributed/metadata_utility.h

@@ -400,7 +400,7 @@ extern bool ShouldPropagateAnyObject(List *addresses);
 /* Remaining metadata utility functions  */
 extern Oid TableOwnerOid(Oid relationId);
 extern char * TableOwner(Oid relationId);
-extern void EnsureTablePermissions(Oid relationId, AclMode mode);
+extern void EnsureTablePermissions(Oid relationId, AclMode mode, AclMaskHow mask);
 extern void EnsureTableOwner(Oid relationId);
 extern void EnsureHashDistributedTable(Oid relationId);
 extern void EnsureHashOrSingleShardDistributedTable(Oid relationId);