Adjust lo_open() so that specifying INV_READ without INV_WRITE creates

a descriptor that uses the current transaction snapshot, rather than SnapshotNow as it did before (and still does if INV_WRITE is set). This means pg_dump will now dump a consistent snapshot of large object contents, as it never could do before. Also, add a lo_create() function that is similar to lo_creat() but allows the desired OID of the large object to be specified. This will simplify pg_restore considerably (but I'll fix that in a separate commit).
2025-09-02 04:21:28 +03:00 · 2005-06-13 02:26:53 +00:00
parent f52a34229b
commit a2fb7b8a1f
11 changed files with 265 additions and 96 deletions
--- a/src/backend/libpq/be-fsstubs.c
+++ b/src/backend/libpq/be-fsstubs.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/libpq/be-fsstubs.c,v 1.77 2004/12/31 21:59:50 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/backend/libpq/be-fsstubs.c,v 1.78 2005/06/13 02:26:48 tgl Exp $
 *
 * NOTES
 *	  This should be moved to a more appropriate place.  It is here
@@ -195,6 +195,12 @@ lo_write(int fd, char *buf, int len)
 		return -1;
 	}

+	if ((cookies[fd]->flags & IFS_WRLOCK) == 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("large object descriptor %d was not opened for writing",
+						fd)));
+
 	Assert(fscxt != NULL);
 	currentContext = MemoryContextSwitchTo(fscxt);

@@ -236,26 +242,33 @@ lo_lseek(PG_FUNCTION_ARGS)
 Datum
 lo_creat(PG_FUNCTION_ARGS)
 {
-	int32		mode = PG_GETARG_INT32(0);
-	LargeObjectDesc *lobjDesc;
-	MemoryContext currentContext;
 	Oid			lobjId;
+	MemoryContext currentContext;

+	/* do we actually need fscxt for this? */
 	CreateFSContext();

 	currentContext = MemoryContextSwitchTo(fscxt);

-	lobjDesc = inv_create(mode);
+	lobjId = inv_create(InvalidOid);

-	if (lobjDesc == NULL)
-	{
-		MemoryContextSwitchTo(currentContext);
-		PG_RETURN_OID(InvalidOid);
-	}
+	MemoryContextSwitchTo(currentContext);

-	lobjId = lobjDesc->id;
+	PG_RETURN_OID(lobjId);
+}

-	inv_close(lobjDesc);
+Datum
+lo_create(PG_FUNCTION_ARGS)
+{
+	Oid			lobjId = PG_GETARG_OID(0);
+	MemoryContext currentContext;
+
+	/* do we actually need fscxt for this? */
+	CreateFSContext();
+
+	currentContext = MemoryContextSwitchTo(fscxt);
+
+	lobjId = inv_create(lobjId);

 	MemoryContextSwitchTo(currentContext);

@@ -403,12 +416,13 @@ lo_import(PG_FUNCTION_ARGS)
 	/*
 	 * create an inversion object
 	 */
-	lobj = inv_create(INV_READ | INV_WRITE);
-	lobjOid = lobj->id;
+	lobjOid = inv_create(InvalidOid);

 	/*
-	 * read in from the filesystem and write to the inversion file
+	 * read in from the filesystem and write to the inversion object
 	 */
+	lobj = inv_open(lobjOid, INV_WRITE);
+
 	while ((nbytes = FileRead(fd, buf, BUFSIZE)) > 0)
 	{
 		tmp = inv_write(lobj, buf, nbytes);
@@ -421,8 +435,8 @@ lo_import(PG_FUNCTION_ARGS)
 				 errmsg("could not read server file \"%s\": %m",
 						fnamebuf)));

-	FileClose(fd);
 	inv_close(lobj);
+	FileClose(fd);

 	PG_RETURN_OID(lobjOid);
 }
--- a/src/backend/storage/large_object/inv_api.c
+++ b/src/backend/storage/large_object/inv_api.c
@@ -9,7 +9,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/storage/large_object/inv_api.c,v 1.110 2005/04/14 20:03:25 tgl Exp $
+ *	  $PostgreSQL: pgsql/src/backend/storage/large_object/inv_api.c,v 1.111 2005/06/13 02:26:49 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -114,6 +114,42 @@ close_lo_relation(bool isCommit)
 }


+/*
+ * Same as pg_largeobject.c's LargeObjectExists(), except snapshot to
+ * read with can be specified.
+ */
+static bool
+myLargeObjectExists(Oid loid, Snapshot snapshot)
+{
+	bool		retval = false;
+	Relation	pg_largeobject;
+	ScanKeyData skey[1];
+	SysScanDesc sd;
+
+	/*
+	 * See if we can find any tuples belonging to the specified LO
+	 */
+	ScanKeyInit(&skey[0],
+				Anum_pg_largeobject_loid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(loid));
+
+	pg_largeobject = heap_open(LargeObjectRelationId, AccessShareLock);
+
+	sd = systable_beginscan(pg_largeobject, LargeObjectLOidPNIndexId, true,
+							snapshot, 1, skey);
+
+	if (systable_getnext(sd) != NULL)
+		retval = true;
+
+	systable_endscan(sd);
+
+	heap_close(pg_largeobject, AccessShareLock);
+
+	return retval;
+}
+
+
 static int32
 getbytealen(bytea *data)
 {
@@ -125,58 +161,44 @@ getbytealen(bytea *data)


 /*
- *	inv_create -- create a new large object.
+ *	inv_create -- create a new large object
 *
- *		Arguments:
- *		  flags
+ *	Arguments:
+ *	  lobjId - OID to use for new large object, or InvalidOid to pick one
 *
- *		Returns:
- *		  large object descriptor, appropriately filled in.
+ *	Returns:
+ *	  OID of new object
+ *
+ * If lobjId is not InvalidOid, then an error occurs if the OID is already
+ * in use.
 */
-LargeObjectDesc *
-inv_create(int flags)
+Oid
+inv_create(Oid lobjId)
 {
-	Oid			file_oid;
-	LargeObjectDesc *retval;
-
 	/*
-	 * Allocate an OID to be the LO's identifier.
+	 * Allocate an OID to be the LO's identifier, unless we were told
+	 * what to use.  In event of collision with an existing ID, loop
+	 * to find a free one.
 	 */
-	file_oid = newoid();
-
-	/* Check for duplicate (shouldn't happen) */
-	if (LargeObjectExists(file_oid))
-		elog(ERROR, "large object %u already exists", file_oid);
+	if (!OidIsValid(lobjId))
+	{
+		do {
+			lobjId = newoid();
+		} while (LargeObjectExists(lobjId));
+	}

 	/*
 	 * Create the LO by writing an empty first page for it in
-	 * pg_largeobject
+	 * pg_largeobject (will fail if duplicate)
 	 */
-	LargeObjectCreate(file_oid);
+	LargeObjectCreate(lobjId);

 	/*
-	 * Advance command counter so that new tuple will be seen by later
-	 * large-object operations in this transaction.
+	 * Advance command counter to make new tuple visible to later operations.
 	 */
 	CommandCounterIncrement();

-	/*
-	 * Prepare LargeObjectDesc data structure for accessing LO
-	 */
-	retval = (LargeObjectDesc *) palloc(sizeof(LargeObjectDesc));
-
-	retval->id = file_oid;
-	retval->subid = GetCurrentSubTransactionId();
-	retval->offset = 0;
-
-	if (flags & INV_WRITE)
-		retval->flags = IFS_WRLOCK | IFS_RDLOCK;
-	else if (flags & INV_READ)
-		retval->flags = IFS_RDLOCK;
-	else
-		elog(ERROR, "invalid flags: %d", flags);
-
-	return retval;
+	return lobjId;
 }

 /*
@@ -190,11 +212,6 @@ inv_open(Oid lobjId, int flags)
 {
 	LargeObjectDesc *retval;

-	if (!LargeObjectExists(lobjId))
-		ereport(ERROR,
-				(errcode(ERRCODE_UNDEFINED_OBJECT),
-				 errmsg("large object %u does not exist", lobjId)));
-
 	retval = (LargeObjectDesc *) palloc(sizeof(LargeObjectDesc));

 	retval->id = lobjId;
@@ -202,12 +219,25 @@ inv_open(Oid lobjId, int flags)
 	retval->offset = 0;

 	if (flags & INV_WRITE)
+	{
+		retval->snapshot = SnapshotNow;
 		retval->flags = IFS_WRLOCK | IFS_RDLOCK;
+	}
 	else if (flags & INV_READ)
+	{
+		/* be sure to copy snap into fscxt */
+		retval->snapshot = CopySnapshot(ActiveSnapshot);
 		retval->flags = IFS_RDLOCK;
+	}
 	else
 		elog(ERROR, "invalid flags: %d", flags);

+	/* Can't use LargeObjectExists here because it always uses SnapshotNow */
+	if (!myLargeObjectExists(lobjId, retval->snapshot))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("large object %u does not exist", lobjId)));
+
 	return retval;
 }

@@ -218,6 +248,8 @@ void
 inv_close(LargeObjectDesc *obj_desc)
 {
 	Assert(PointerIsValid(obj_desc));
+	if (obj_desc->snapshot != SnapshotNow)
+		FreeSnapshot(obj_desc->snapshot);
 	pfree(obj_desc);
 }

@@ -268,7 +300,7 @@ inv_getsize(LargeObjectDesc *obj_desc)
 				ObjectIdGetDatum(obj_desc->id));

 	sd = index_beginscan(lo_heap_r, lo_index_r,
-						 SnapshotNow, 1, skey);
+						 obj_desc->snapshot, 1, skey);

 	/*
 	 * Because the pg_largeobject index is on both loid and pageno, but we
@@ -379,7 +411,7 @@ inv_read(LargeObjectDesc *obj_desc, char *buf, int nbytes)
 				Int32GetDatum(pageno));

 	sd = index_beginscan(lo_heap_r, lo_index_r,
-						 SnapshotNow, 2, skey);
+						 obj_desc->snapshot, 2, skey);

 	while ((tuple = index_getnext(sd, ForwardScanDirection)) != NULL)
 	{
@@ -470,6 +502,13 @@ inv_write(LargeObjectDesc *obj_desc, char *buf, int nbytes)
 	Assert(PointerIsValid(obj_desc));
 	Assert(buf != NULL);

+	/* enforce writability because snapshot is probably wrong otherwise */
+	if ((obj_desc->flags & IFS_WRLOCK) == 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("large object %u was not opened for writing",
+						obj_desc->id)));
+
 	if (nbytes <= 0)
 		return 0;

@@ -488,7 +527,7 @@ inv_write(LargeObjectDesc *obj_desc, char *buf, int nbytes)
 				Int32GetDatum(pageno));

 	sd = index_beginscan(lo_heap_r, lo_index_r,
-						 SnapshotNow, 2, skey);
+						 obj_desc->snapshot, 2, skey);

 	oldtuple = NULL;
 	olddata = NULL;
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -37,7 +37,7 @@
 * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
- * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.273 2005/06/07 07:08:34 neilc Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/catversion.h,v 1.274 2005/06/13 02:26:50 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -53,6 +53,6 @@
 */

 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	200506071
+#define CATALOG_VERSION_NO	200506121

 #endif
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -7,7 +7,7 @@
 * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
- * $PostgreSQL: pgsql/src/include/catalog/pg_proc.h,v 1.365 2005/06/09 16:35:09 momjian Exp $
+ * $PostgreSQL: pgsql/src/include/catalog/pg_proc.h,v 1.366 2005/06/13 02:26:50 tgl Exp $
 *
 * NOTES
 *	  The script catalog/genbki.sh reads this file and generates .bki
@@ -1248,6 +1248,8 @@ DATA(insert OID = 956 (  lo_lseek		   PGNSP PGUID 12 f f t f v 3 23 "23 23 23" _
 DESCR("large object seek");
 DATA(insert OID = 957 (  lo_creat		   PGNSP PGUID 12 f f t f v 1 26 "23" _null_ _null_ _null_  lo_creat - _null_ ));
 DESCR("large object create");
+DATA(insert OID = 715 (  lo_create		   PGNSP PGUID 12 f f t f v 1 26 "26" _null_ _null_ _null_  lo_create - _null_ ));
+DESCR("large object create");
 DATA(insert OID = 958 (  lo_tell		   PGNSP PGUID 12 f f t f v 1 23 "23" _null_ _null_ _null_  lo_tell - _null_ ));
 DESCR("large object position");

--- a/src/include/libpq/be-fsstubs.h
+++ b/src/include/libpq/be-fsstubs.h
@@ -7,7 +7,7 @@
 * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
- * $PostgreSQL: pgsql/src/include/libpq/be-fsstubs.h,v 1.23 2004/12/31 22:03:32 pgsql Exp $
+ * $PostgreSQL: pgsql/src/include/libpq/be-fsstubs.h,v 1.24 2005/06/13 02:26:51 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -23,6 +23,7 @@ extern Datum lo_import(PG_FUNCTION_ARGS);
 extern Datum lo_export(PG_FUNCTION_ARGS);

 extern Datum lo_creat(PG_FUNCTION_ARGS);
+extern Datum lo_create(PG_FUNCTION_ARGS);

 extern Datum lo_open(PG_FUNCTION_ARGS);
 extern Datum lo_close(PG_FUNCTION_ARGS);
--- a/src/include/storage/large_object.h
+++ b/src/include/storage/large_object.h
@@ -8,19 +8,22 @@
 * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
- * $PostgreSQL: pgsql/src/include/storage/large_object.h,v 1.31 2004/12/31 22:03:42 pgsql Exp $
+ * $PostgreSQL: pgsql/src/include/storage/large_object.h,v 1.32 2005/06/13 02:26:52 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
 #ifndef LARGE_OBJECT_H
 #define LARGE_OBJECT_H

+#include "utils/tqual.h"
+

 /*----------
 * Data about a currently-open large object.
 *
 * id is the logical OID of the large object
- * subid is the subtransaction that opened the LO (or currently owns it)
+ * snapshot is the snapshot to use for read/write operations
+ * subid is the subtransaction that opened the desc (or currently owns it)
 * offset is the current seek offset within the LO
 * flags contains some flag bits
 *
@@ -32,6 +35,7 @@
 typedef struct LargeObjectDesc
 {
 	Oid			id;				/* LO's identifier */
+	Snapshot	snapshot;		/* snapshot to use */
 	SubTransactionId subid;		/* owning subtransaction ID */
 	uint32		offset;			/* current seek pointer */
 	int			flags;			/* locking info, etc */
@@ -39,6 +43,7 @@ typedef struct LargeObjectDesc
 /* flag bits: */
 #define IFS_RDLOCK		(1 << 0)
 #define IFS_WRLOCK		(1 << 1)
+
 } LargeObjectDesc;


@@ -65,7 +70,7 @@ typedef struct LargeObjectDesc

 /* inversion stuff in inv_api.c */
 extern void close_lo_relation(bool isCommit);
-extern LargeObjectDesc *inv_create(int flags);
+extern Oid	inv_create(Oid lobjId);
 extern LargeObjectDesc *inv_open(Oid lobjId, int flags);
 extern void inv_close(LargeObjectDesc *obj_desc);
 extern int	inv_drop(Oid lobjId);
--- a/src/interfaces/libpq/exports.txt
+++ b/src/interfaces/libpq/exports.txt
@@ -1,4 +1,4 @@
-# $PostgreSQL: pgsql/src/interfaces/libpq/exports.txt,v 1.3 2004/10/30 23:11:26 tgl Exp $
+# $PostgreSQL: pgsql/src/interfaces/libpq/exports.txt,v 1.4 2005/06/13 02:26:53 tgl Exp $
 # Functions to be exported by libpq DLLs
 PQconnectdb               1
 PQsetdbLogin              2
@@ -122,3 +122,4 @@ PQsendPrepare             119
 PQgetCancel               120
 PQfreeCancel              121
 PQcancel                  122
+lo_create                 123
--- a/src/interfaces/libpq/fe-lobj.c
+++ b/src/interfaces/libpq/fe-lobj.c
@@ -8,7 +8,7 @@
 *
 *
 * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/interfaces/libpq/fe-lobj.c,v 1.52 2004/12/31 22:03:50 pgsql Exp $
+ *	  $PostgreSQL: pgsql/src/interfaces/libpq/fe-lobj.c,v 1.53 2005/06/13 02:26:53 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -266,12 +266,11 @@ lo_lseek(PGconn *conn, int fd, int offset, int whence)
 /*
 * lo_creat
 *	  create a new large object
- * the mode is a bitmask describing different attributes of the new object
+ * the mode is ignored (once upon a time it had a use)
 *
 * returns the oid of the large object created or
 * InvalidOid upon failure
 */
-
 Oid
 lo_creat(PGconn *conn, int mode)
 {
@@ -303,6 +302,53 @@ lo_creat(PGconn *conn, int mode)
 	}
 }

+/*
+ * lo_create
+ *	  create a new large object
+ * if lobjId isn't InvalidOid, it specifies the OID to (attempt to) create
+ *
+ * returns the oid of the large object created or
+ * InvalidOid upon failure
+ */
+Oid
+lo_create(PGconn *conn, Oid lobjId)
+{
+	PQArgBlock	argv[1];
+	PGresult   *res;
+	int			retval;
+	int			result_len;
+
+	if (conn->lobjfuncs == NULL)
+	{
+		if (lo_initialize(conn) < 0)
+			return InvalidOid;
+	}
+
+	/* Must check this on-the-fly because it's not there pre-8.1 */
+	if (conn->lobjfuncs->fn_lo_create == 0)
+	{
+		printfPQExpBuffer(&conn->errorMessage,
+		   libpq_gettext("cannot determine OID of function lo_create\n"));
+		return InvalidOid;
+	}
+
+	argv[0].isint = 1;
+	argv[0].len = 4;
+	argv[0].u.integer = lobjId;
+	res = PQfn(conn, conn->lobjfuncs->fn_lo_create,
+			   &retval, &result_len, 1, argv, 1);
+	if (PQresultStatus(res) == PGRES_COMMAND_OK)
+	{
+		PQclear(res);
+		return (Oid) retval;
+	}
+	else
+	{
+		PQclear(res);
+		return InvalidOid;
+	}
+}
+

 /*
 * lo_tell
@@ -560,7 +606,8 @@ lo_initialize(PGconn *conn)

 	/*
 	 * Execute the query to get all the functions at once.	In 7.3 and
-	 * later we need to be schema-safe.
+	 * later we need to be schema-safe.  lo_create only exists in 8.1
+	 * and up.
 	 */
 	if (conn->sversion >= 70300)
 		query = "select proname, oid from pg_catalog.pg_proc "
@@ -568,6 +615,7 @@ lo_initialize(PGconn *conn)
 			"'lo_open', "
 			"'lo_close', "
 			"'lo_creat', "
+			"'lo_create', "
 			"'lo_unlink', "
 			"'lo_lseek', "
 			"'lo_tell', "
@@ -615,6 +663,8 @@ lo_initialize(PGconn *conn)
 			lobjfuncs->fn_lo_close = foid;
 		else if (!strcmp(fname, "lo_creat"))
 			lobjfuncs->fn_lo_creat = foid;
+		else if (!strcmp(fname, "lo_create"))
+			lobjfuncs->fn_lo_create = foid;
 		else if (!strcmp(fname, "lo_unlink"))
 			lobjfuncs->fn_lo_unlink = foid;
 		else if (!strcmp(fname, "lo_lseek"))
@@ -631,7 +681,7 @@ lo_initialize(PGconn *conn)

 	/*
 	 * Finally check that we really got all large object interface
-	 * functions.
+	 * functions --- except lo_create, which may not exist.
 	 */
 	if (lobjfuncs->fn_lo_open == 0)
 	{
--- a/src/interfaces/libpq/libpq-fe.h
+++ b/src/interfaces/libpq/libpq-fe.h
@@ -7,7 +7,7 @@
 * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
- * $PostgreSQL: pgsql/src/interfaces/libpq/libpq-fe.h,v 1.117 2005/06/09 20:01:16 tgl Exp $
+ * $PostgreSQL: pgsql/src/interfaces/libpq/libpq-fe.h,v 1.118 2005/06/13 02:26:53 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -480,6 +480,7 @@ extern int	lo_read(PGconn *conn, int fd, char *buf, size_t len);
 extern int	lo_write(PGconn *conn, int fd, char *buf, size_t len);
 extern int	lo_lseek(PGconn *conn, int fd, int offset, int whence);
 extern Oid	lo_creat(PGconn *conn, int mode);
+extern Oid	lo_create(PGconn *conn, Oid lobjId);
 extern int	lo_tell(PGconn *conn, int fd);
 extern int	lo_unlink(PGconn *conn, Oid lobjId);
 extern Oid	lo_import(PGconn *conn, const char *filename);
--- a/src/interfaces/libpq/libpq-int.h
+++ b/src/interfaces/libpq/libpq-int.h
@@ -12,7 +12,7 @@
 * Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
- * $PostgreSQL: pgsql/src/interfaces/libpq/libpq-int.h,v 1.102 2005/06/12 00:00:21 neilc Exp $
+ * $PostgreSQL: pgsql/src/interfaces/libpq/libpq-int.h,v 1.103 2005/06/13 02:26:53 tgl Exp $
 *
 *-------------------------------------------------------------------------
 */
@@ -229,6 +229,7 @@ typedef struct pgLobjfuncs
 	Oid			fn_lo_open;		/* OID of backend function lo_open		*/
 	Oid			fn_lo_close;	/* OID of backend function lo_close		*/
 	Oid			fn_lo_creat;	/* OID of backend function lo_creat		*/
+	Oid			fn_lo_create;	/* OID of backend function lo_create	*/
 	Oid			fn_lo_unlink;	/* OID of backend function lo_unlink	*/
 	Oid			fn_lo_lseek;	/* OID of backend function lo_lseek		*/
 	Oid			fn_lo_tell;		/* OID of backend function lo_tell		*/