Provide multi-block smgrprefetch().

Previously smgrprefetch() could issue POSIX_FADV_WILLNEED advice for a
single block at a time.  Add an nblocks argument so that we can do the
same for a range of blocks.  This usually produces a single system call,
but might need to loop if it crosses a segment boundary.  Initially it
is only called with nblocks == 1, but proposed patches will make wider
calls.

Reviewed-by: Heikki Linnakangas <hlinnaka@iki.fi> (earlier version)
Discussion: https://postgr.es/m/CA+hUKGJkOiOCa+mag4BF+zHo7qo=o9CFheB8=g6uT5TUm2gkvA@mail.gmail.com

src/backend/catalog/aclchk.c

@@ -4103,9 +4103,14 @@ object_ownercheck(Oid classid, Oid objectid, Oid roleid)
 	if (superuser_arg(roleid))
 		return true;
 
+	/* For large objects, the catalog to consult is pg_largeobject_metadata */
+	if (classid == LargeObjectRelationId)
+		classid = LargeObjectMetadataRelationId;
+
 	cacheid = get_object_catcache_oid(classid);
 	if (cacheid != -1)
 	{
+		/* we can get the object's tuple from the syscache */
 		HeapTuple	tuple;
 
 		tuple = SearchSysCache1(cacheid, ObjectIdGetDatum(objectid));
@@ -4122,7 +4127,6 @@ object_ownercheck(Oid classid, Oid objectid, Oid roleid)
 	else
 	{
 		/* for catalogs without an appropriate syscache */
-
 		Relation	rel;
 		ScanKeyData entry[1];
 		SysScanDesc scan;
@@ -4442,9 +4446,9 @@ recordExtObjInitPriv(Oid objoid, Oid classoid)
 
 		ReleaseSysCache(tuple);
 	}
-	/* pg_largeobject_metadata */
-	else if (classoid == LargeObjectMetadataRelationId)
+	else if (classoid == LargeObjectRelationId)
 	{
+		/* For large objects, we must consult pg_largeobject_metadata */
 		Datum		aclDatum;
 		bool		isNull;
 		HeapTuple	tuple;

src/backend/catalog/pg_shdepend.c

@@ -1614,20 +1614,9 @@ shdepReassignOwned(List *roleids, Oid newrole)
 				case DatabaseRelationId:
 				case TSConfigRelationId:
 				case TSDictionaryRelationId:
-					{
-						Oid			classId = sdepForm->classid;
-						Relation	catalog;
-
-						if (classId == LargeObjectRelationId)
-							classId = LargeObjectMetadataRelationId;
-
-						catalog = table_open(classId, RowExclusiveLock);
-
-						AlterObjectOwner_internal(catalog, sdepForm->objid,
-												  newrole);
-
-						table_close(catalog, NoLock);
-					}
+					AlterObjectOwner_internal(sdepForm->classid,
+											  sdepForm->objid,
+											  newrole);
 					break;
 
 				default:

src/backend/commands/alter.c

@@ -918,9 +918,7 @@ ExecAlterOwnerStmt(AlterOwnerStmt *stmt)
 		case OBJECT_TSDICTIONARY:
 		case OBJECT_TSCONFIGURATION:
 			{
-				Relation	catalog;
 				Relation	relation;
-				Oid			classId;
 				ObjectAddress address;
 
 				address = get_object_address(stmt->objectType,
@@ -929,20 +927,9 @@ ExecAlterOwnerStmt(AlterOwnerStmt *stmt)
 											 AccessExclusiveLock,
 											 false);
 				Assert(relation == NULL);
-				classId = address.classId;
 
-				/*
-				 * XXX - get_object_address returns Oid of pg_largeobject
-				 * catalog for OBJECT_LARGEOBJECT because of historical
-				 * reasons.  Fix up it here.
-				 */
-				if (classId == LargeObjectRelationId)
-					classId = LargeObjectMetadataRelationId;
-
-				catalog = table_open(classId, RowExclusiveLock);
-
-				AlterObjectOwner_internal(catalog, address.objectId, newowner);
-				table_close(catalog, RowExclusiveLock);
+				AlterObjectOwner_internal(address.classId, address.objectId,
+										  newowner);
 
 				return address;
 			}
@@ -960,25 +947,32 @@ ExecAlterOwnerStmt(AlterOwnerStmt *stmt)
  * cases (won't work for tables, nor other cases where we need to do more than
  * change the ownership column of a single catalog entry).
  *
- * rel: catalog relation containing object (RowExclusiveLock'd by caller)
+ * classId: OID of catalog containing object
  * objectId: OID of object to change the ownership of
  * new_ownerId: OID of new object owner
+ *
+ * This will work on large objects, but we have to beware of the fact that
+ * classId isn't the OID of the catalog to modify in that case.
  */
 void
-AlterObjectOwner_internal(Relation rel, Oid objectId, Oid new_ownerId)
+AlterObjectOwner_internal(Oid classId, Oid objectId, Oid new_ownerId)
 {
-	Oid			classId = RelationGetRelid(rel);
-	AttrNumber	Anum_oid = get_object_attnum_oid(classId);
-	AttrNumber	Anum_owner = get_object_attnum_owner(classId);
-	AttrNumber	Anum_namespace = get_object_attnum_namespace(classId);
-	AttrNumber	Anum_acl = get_object_attnum_acl(classId);
-	AttrNumber	Anum_name = get_object_attnum_name(classId);
+	/* For large objects, the catalog to modify is pg_largeobject_metadata */
+	Oid			catalogId = (classId == LargeObjectRelationId) ? LargeObjectMetadataRelationId : classId;
+	AttrNumber	Anum_oid = get_object_attnum_oid(catalogId);
+	AttrNumber	Anum_owner = get_object_attnum_owner(catalogId);
+	AttrNumber	Anum_namespace = get_object_attnum_namespace(catalogId);
+	AttrNumber	Anum_acl = get_object_attnum_acl(catalogId);
+	AttrNumber	Anum_name = get_object_attnum_name(catalogId);
+	Relation	rel;
 	HeapTuple	oldtup;
 	Datum		datum;
 	bool		isnull;
 	Oid			old_ownerId;
 	Oid			namespaceId = InvalidOid;
 
+	rel = table_open(catalogId, RowExclusiveLock);
+
 	oldtup = get_catalog_object_by_oid(rel, Anum_oid, objectId);
 	if (oldtup == NULL)
 		elog(ERROR, "cache lookup failed for object %u of catalog \"%s\"",
@@ -1026,7 +1020,8 @@ AlterObjectOwner_internal(Relation rel, Oid objectId, Oid new_ownerId)
 					snprintf(namebuf, sizeof(namebuf), "%u", objectId);
 					objname = namebuf;
 				}
-				aclcheck_error(ACLCHECK_NOT_OWNER, get_object_type(classId, objectId),
+				aclcheck_error(ACLCHECK_NOT_OWNER,
+							   get_object_type(catalogId, objectId),
 							   objname);
 			}
 			/* Must be able to become new owner */
@@ -1079,8 +1074,6 @@ AlterObjectOwner_internal(Relation rel, Oid objectId, Oid new_ownerId)
 		CatalogTupleUpdate(rel, &newtup->t_self, newtup);
 
 		/* Update owner dependency reference */
-		if (classId == LargeObjectMetadataRelationId)
-			classId = LargeObjectRelationId;
 		changeDependencyOnOwner(classId, objectId, new_ownerId);
 
 		/* Release memory */
@@ -1089,5 +1082,8 @@ AlterObjectOwner_internal(Relation rel, Oid objectId, Oid new_ownerId)
 		pfree(replaces);
 	}
 
+	/* Note the post-alter hook gets classId not catalogId */
 	InvokeObjectPostAlterHook(classId, objectId, 0);
+
+	table_close(rel, RowExclusiveLock);
 }

src/backend/libpq/be-fsstubs.c

@@ -43,7 +43,7 @@
 #include <unistd.h>
 
 #include "access/xact.h"
-#include "catalog/pg_largeobject_metadata.h"
+#include "catalog/pg_largeobject.h"
 #include "libpq/be-fsstubs.h"
 #include "libpq/libpq-fs.h"
 #include "miscadmin.h"
@@ -323,7 +323,7 @@ be_lo_unlink(PG_FUNCTION_ARGS)
 	 * relevant FDs.
 	 */
 	if (!lo_compat_privileges &&
-		!object_ownercheck(LargeObjectMetadataRelationId, lobjId, GetUserId()))
+		!object_ownercheck(LargeObjectRelationId, lobjId, GetUserId()))
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 				 errmsg("must be owner of large object %u", lobjId)));

src/backend/storage/buffer/bufmgr.c

@@ -569,7 +569,7 @@ PrefetchSharedBuffer(SMgrRelation smgr_reln,
 		 * recovery if the relation file doesn't exist.
 		 */
 		if ((io_direct_flags & IO_DIRECT_DATA) == 0 &&
-			smgrprefetch(smgr_reln, forkNum, blockNum))
+			smgrprefetch(smgr_reln, forkNum, blockNum, 1))
 		{
 			result.initiated_io = true;
 		}

src/backend/storage/buffer/localbuf.c

@@ -94,7 +94,7 @@ PrefetchLocalBuffer(SMgrRelation smgr, ForkNumber forkNum,
 #ifdef USE_PREFETCH
 		/* Not in buffers, so initiate prefetch */
 		if ((io_direct_flags & IO_DIRECT_DATA) == 0 &&
-			smgrprefetch(smgr, forkNum, blockNum))
+			smgrprefetch(smgr, forkNum, blockNum, 1))
 		{
 			result.initiated_io = true;
 		}

src/backend/storage/large_object/inv_api.c

@@ -221,11 +221,10 @@ inv_create(Oid lobjId)
 	/*
 	 * dependency on the owner of largeobject
 	 *
-	 * The reason why we use LargeObjectRelationId instead of
-	 * LargeObjectMetadataRelationId here is to provide backward compatibility
-	 * to the applications which utilize a knowledge about internal layout of
-	 * system catalogs. OID of pg_largeobject_metadata and loid of
-	 * pg_largeobject are same value, so there are no actual differences here.
+	 * Note that LO dependencies are recorded using classId
+	 * LargeObjectRelationId for backwards-compatibility reasons.  Using
+	 * LargeObjectMetadataRelationId instead would simplify matters for the
+	 * backend, but it'd complicate pg_dump and possibly break other clients.
 	 */
 	recordDependencyOnOwner(LargeObjectRelationId,
 							lobjId_new, GetUserId());

src/backend/storage/smgr/md.c

@@ -710,27 +710,44 @@ mdclose(SMgrRelation reln, ForkNumber forknum)
 }
 
 /*
- * mdprefetch() -- Initiate asynchronous read of the specified block of a relation
+ * mdprefetch() -- Initiate asynchronous read of the specified blocks of a relation
  */
 bool
-mdprefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
+mdprefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+		   int nblocks)
 {
 #ifdef USE_PREFETCH
-	off_t		seekpos;
-	MdfdVec    *v;
 
 	Assert((io_direct_flags & IO_DIRECT_DATA) == 0);
 
-	v = _mdfd_getseg(reln, forknum, blocknum, false,
-					 InRecovery ? EXTENSION_RETURN_NULL : EXTENSION_FAIL);
-	if (v == NULL)
+	if ((uint64) blocknum + nblocks > (uint64) MaxBlockNumber + 1)
 		return false;
 
-	seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+	while (nblocks > 0)
+	{
+		off_t		seekpos;
+		MdfdVec    *v;
+		int			nblocks_this_segment;
 
-	Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+		v = _mdfd_getseg(reln, forknum, blocknum, false,
+						 InRecovery ? EXTENSION_RETURN_NULL : EXTENSION_FAIL);
+		if (v == NULL)
+			return false;
 
-	(void) FilePrefetch(v->mdfd_vfd, seekpos, BLCKSZ, WAIT_EVENT_DATA_FILE_PREFETCH);
+		seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+
+		Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+
+		nblocks_this_segment =
+			Min(nblocks,
+				RELSEG_SIZE - (blocknum % ((BlockNumber) RELSEG_SIZE)));
+
+		(void) FilePrefetch(v->mdfd_vfd, seekpos, BLCKSZ * nblocks_this_segment,
+							WAIT_EVENT_DATA_FILE_PREFETCH);
+
+		blocknum += nblocks_this_segment;
+		nblocks -= nblocks_this_segment;
+	}
 #endif							/* USE_PREFETCH */
 
 	return true;

src/backend/storage/smgr/smgr.c

@@ -54,7 +54,7 @@ typedef struct f_smgr
 	void		(*smgr_zeroextend) (SMgrRelation reln, ForkNumber forknum,
 									BlockNumber blocknum, int nblocks, bool skipFsync);
 	bool		(*smgr_prefetch) (SMgrRelation reln, ForkNumber forknum,
-								  BlockNumber blocknum);
+								  BlockNumber blocknum, int nblocks);
 	void		(*smgr_read) (SMgrRelation reln, ForkNumber forknum,
 							  BlockNumber blocknum, void *buffer);
 	void		(*smgr_write) (SMgrRelation reln, ForkNumber forknum,
@@ -547,9 +547,10 @@ smgrzeroextend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
  * record).
  */
 bool
-smgrprefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum)
+smgrprefetch(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+			 int nblocks)
 {
-	return smgrsw[reln->smgr_which].smgr_prefetch(reln, forknum, blocknum);
+	return smgrsw[reln->smgr_which].smgr_prefetch(reln, forknum, blocknum, nblocks);
 }
 
 /*

src/include/commands/alter.h

@@ -17,7 +17,6 @@
 #include "catalog/dependency.h"
 #include "catalog/objectaddress.h"
 #include "nodes/parsenodes.h"
-#include "utils/relcache.h"
 
 extern ObjectAddress ExecRenameStmt(RenameStmt *stmt);
 
@@ -29,7 +28,7 @@ extern Oid	AlterObjectNamespace_oid(Oid classId, Oid objid, Oid nspOid,
 									 ObjectAddresses *objsMoved);
 
 extern ObjectAddress ExecAlterOwnerStmt(AlterOwnerStmt *stmt);
-extern void AlterObjectOwner_internal(Relation rel, Oid objectId,
+extern void AlterObjectOwner_internal(Oid classId, Oid objectId,
 									  Oid new_ownerId);
 
 #endif							/* ALTER_H */

src/include/storage/md.h

@@ -31,7 +31,7 @@ extern void mdextend(SMgrRelation reln, ForkNumber forknum,
 extern void mdzeroextend(SMgrRelation reln, ForkNumber forknum,
 						 BlockNumber blocknum, int nblocks, bool skipFsync);
 extern bool mdprefetch(SMgrRelation reln, ForkNumber forknum,
-					   BlockNumber blocknum);
+					   BlockNumber blocknum, int nblocks);
 extern void mdread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 				   void *buffer);
 extern void mdwrite(SMgrRelation reln, ForkNumber forknum,

src/include/storage/smgr.h

@@ -95,7 +95,7 @@ extern void smgrextend(SMgrRelation reln, ForkNumber forknum,
 extern void smgrzeroextend(SMgrRelation reln, ForkNumber forknum,
 						   BlockNumber blocknum, int nblocks, bool skipFsync);
 extern bool smgrprefetch(SMgrRelation reln, ForkNumber forknum,
-						 BlockNumber blocknum);
+						 BlockNumber blocknum, int nblocks);
 extern void smgrread(SMgrRelation reln, ForkNumber forknum,
 					 BlockNumber blocknum, void *buffer);
 extern void smgrwrite(SMgrRelation reln, ForkNumber forknum,

src/test/regress/expected/largeobject.out

@@ -6,7 +6,7 @@
 \getenv abs_builddir PG_ABS_BUILDDIR
 -- ensure consistent test output regardless of the default bytea format
 SET bytea_output TO escape;
--- Test ALTER LARGE OBJECT OWNER, GRANT, COMMENT
+-- Test ALTER LARGE OBJECT OWNER
 CREATE ROLE regress_lo_user;
 SELECT lo_create(42);
  lo_create 
@@ -15,8 +15,11 @@ SELECT lo_create(42);
 (1 row)
 
 ALTER LARGE OBJECT 42 OWNER TO regress_lo_user;
+-- Test GRANT, COMMENT as non-superuser
+SET SESSION AUTHORIZATION regress_lo_user;
 GRANT SELECT ON LARGE OBJECT 42 TO public;
 COMMENT ON LARGE OBJECT 42 IS 'the ultimate answer';
+RESET SESSION AUTHORIZATION;
 -- Test psql's \lo_list et al (we assume no other LOs exist yet)
 \lo_list
                Large objects

src/test/regress/expected/largeobject_1.out

@@ -6,7 +6,7 @@
 \getenv abs_builddir PG_ABS_BUILDDIR
 -- ensure consistent test output regardless of the default bytea format
 SET bytea_output TO escape;
--- Test ALTER LARGE OBJECT OWNER, GRANT, COMMENT
+-- Test ALTER LARGE OBJECT OWNER
 CREATE ROLE regress_lo_user;
 SELECT lo_create(42);
  lo_create 
@@ -15,8 +15,11 @@ SELECT lo_create(42);
 (1 row)
 
 ALTER LARGE OBJECT 42 OWNER TO regress_lo_user;
+-- Test GRANT, COMMENT as non-superuser
+SET SESSION AUTHORIZATION regress_lo_user;
 GRANT SELECT ON LARGE OBJECT 42 TO public;
 COMMENT ON LARGE OBJECT 42 IS 'the ultimate answer';
+RESET SESSION AUTHORIZATION;
 -- Test psql's \lo_list et al (we assume no other LOs exist yet)
 \lo_list
                Large objects

src/test/regress/sql/largeobject.sql

@@ -9,13 +9,19 @@
 -- ensure consistent test output regardless of the default bytea format
 SET bytea_output TO escape;
 
--- Test ALTER LARGE OBJECT OWNER, GRANT, COMMENT
+-- Test ALTER LARGE OBJECT OWNER
 CREATE ROLE regress_lo_user;
 SELECT lo_create(42);
 ALTER LARGE OBJECT 42 OWNER TO regress_lo_user;
+
+-- Test GRANT, COMMENT as non-superuser
+SET SESSION AUTHORIZATION regress_lo_user;
+
 GRANT SELECT ON LARGE OBJECT 42 TO public;
 COMMENT ON LARGE OBJECT 42 IS 'the ultimate answer';
 
+RESET SESSION AUTHORIZATION;
+
 -- Test psql's \lo_list et al (we assume no other LOs exist yet)
 \lo_list
 \lo_list+