Major overhaul of large-object implementation, by Denis Perchine with

kibitzing from Tom Lane.  Large objects are now all stored in a single
system relation "pg_largeobject" --- no more xinv or xinx files, no more
relkind 'l'.  This should offer substantial performance improvement for
large numbers of LOs, since there won't be directory bloat anymore.
It'll also fix problems like running out of locktable space when you
access thousands of LOs in one transaction.
Also clean up cruft in read/write routines.  LOs with "holes" in them
(never-written byte ranges) now work just like Unix files with holes do:
a hole reads as zeroes but doesn't occupy storage space.
INITDB forced!

src/backend/catalog/Makefile

@@ -2,7 +2,7 @@
 #
 # Makefile for catalog
 #
-# $Header: /cvsroot/pgsql/src/backend/catalog/Makefile,v 1.30 2000/10/22 05:27:10 momjian Exp $
+# $Header: /cvsroot/pgsql/src/backend/catalog/Makefile,v 1.31 2000/10/24 01:38:23 tgl Exp $
 #
 #-------------------------------------------------------------------------
 
@@ -11,7 +11,8 @@ top_builddir = ../../..
 include $(top_builddir)/src/Makefile.global
 
 OBJS = catalog.o heap.o index.o indexing.o aclchk.o \
-       pg_aggregate.o pg_operator.o pg_proc.o pg_type.o
+       pg_aggregate.o pg_largeobject.o pg_operator.o pg_proc.o \
+       pg_type.o
 
 BKIFILES = global.bki template1.bki global.description template1.description
 
@@ -29,7 +30,7 @@ TEMPLATE1_BKI_SRCS := $(addprefix $(top_srcdir)/src/include/catalog/,\
 	pg_proc.h pg_type.h pg_attribute.h pg_class.h \
 	pg_inherits.h pg_index.h pg_statistic.h \
 	pg_operator.h pg_opclass.h pg_am.h pg_amop.h pg_amproc.h \
-	pg_language.h \
+	pg_language.h pg_largeobject.h \
 	pg_aggregate.h pg_ipl.h pg_inheritproc.h \
 	pg_rewrite.h pg_listener.h pg_description.h indexing.h \
     )

src/backend/catalog/indexing.c

@@ -9,7 +9,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/catalog/indexing.c,v 1.71 2000/10/22 05:27:10 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/catalog/indexing.c,v 1.72 2000/10/24 01:38:22 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -51,6 +51,8 @@ char	   *Name_pg_inherits_indices[Num_pg_inherits_indices] =
 {InheritsRelidSeqnoIndex};
 char	   *Name_pg_language_indices[Num_pg_language_indices] =
 {LanguageOidIndex, LanguageNameIndex};
+char	   *Name_pg_largeobject_indices[Num_pg_largeobject_indices] =
+{LargeObjectLOidPNIndex};
 char	   *Name_pg_listener_indices[Num_pg_listener_indices] =
 {ListenerPidRelnameIndex};
 char	   *Name_pg_opclass_indices[Num_pg_opclass_indices] =

src/backend/catalog/pg_largeobject.c

@@ -0,0 +1,184 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_largeobject.c
+ *	  routines to support manipulation of the pg_largeobject relation
+ *
+ * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  $Header: /cvsroot/pgsql/src/backend/catalog/pg_largeobject.c,v 1.5 2000/10/24 01:38:23 tgl Exp $
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/genam.h"
+#include "access/heapam.h"
+#include "catalog/catname.h"
+#include "catalog/indexing.h"
+#include "catalog/pg_largeobject.h"
+#include "miscadmin.h"
+#include "utils/builtins.h"
+#include "utils/fmgroids.h"
+
+
+/*
+ * Create a large object having the given LO identifier.
+ *
+ * We do this by inserting an empty first page, so that the object will
+ * appear to exist with size 0.  Note that the unique index will reject
+ * an attempt to create a duplicate page.
+ *
+ * Return value is OID assigned to the page tuple (any use in it?)
+ */
+Oid
+LargeObjectCreate(Oid loid)
+{
+	Oid			retval;
+	Relation	pg_largeobject;
+	HeapTuple	ntup;
+	Relation	idescs[Num_pg_largeobject_indices];
+	Datum		values[Natts_pg_largeobject];
+	char		nulls[Natts_pg_largeobject];
+	int			i;
+
+	pg_largeobject = heap_openr(LargeObjectRelationName, RowExclusiveLock);
+
+	/*
+	 * Form new tuple
+	 */
+	for (i = 0; i < Natts_pg_largeobject; i++)
+	{
+		values[i] = (Datum)NULL;
+		nulls[i] = ' ';
+	}
+
+	i = 0;
+	values[i++] = ObjectIdGetDatum(loid);
+	values[i++] = Int32GetDatum(0);
+	values[i++] = DirectFunctionCall1(byteain,
+									  CStringGetDatum(""));
+	
+	ntup = heap_formtuple(pg_largeobject->rd_att, values, nulls);
+
+	/*
+	 * Insert it
+	 */
+	retval = heap_insert(pg_largeobject, ntup);
+
+	/*
+	 * Update indices
+	 */
+	if (!IsIgnoringSystemIndexes())
+	{
+		CatalogOpenIndices(Num_pg_largeobject_indices, Name_pg_largeobject_indices, idescs);
+		CatalogIndexInsert(idescs, Num_pg_largeobject_indices, pg_largeobject, ntup);
+		CatalogCloseIndices(Num_pg_largeobject_indices, idescs);
+	}
+	
+	heap_close(pg_largeobject, RowExclusiveLock);
+
+	heap_freetuple(ntup);
+
+	return retval;
+}
+
+void
+LargeObjectDrop(Oid loid)
+{
+	bool		found = false;
+	Relation	pg_largeobject;
+	Relation	pg_lo_idx;
+	ScanKeyData	skey[1];
+	IndexScanDesc sd;
+	RetrieveIndexResult	indexRes;
+	HeapTupleData tuple;
+	Buffer		buffer;
+
+	ScanKeyEntryInitialize(&skey[0],
+						   (bits16) 0x0,
+						   (AttrNumber) 1,
+						   (RegProcedure) F_OIDEQ,
+						   ObjectIdGetDatum(loid));
+
+	pg_largeobject = heap_openr(LargeObjectRelationName, RowShareLock);
+	pg_lo_idx = index_openr(LargeObjectLOidPNIndex);
+
+	sd = index_beginscan(pg_lo_idx, false, 1, skey);
+
+	tuple.t_datamcxt = CurrentMemoryContext;
+	tuple.t_data = NULL;
+
+	while ((indexRes = index_getnext(sd, ForwardScanDirection)))
+	{
+		tuple.t_self = indexRes->heap_iptr;
+		heap_fetch(pg_largeobject, SnapshotNow, &tuple, &buffer);
+		pfree(indexRes);
+		if (tuple.t_data != NULL)
+		{
+			heap_delete(pg_largeobject, &tuple.t_self, NULL);
+			ReleaseBuffer(buffer);
+			found = true;
+		}
+	}
+
+	index_endscan(sd);
+
+	index_close(pg_lo_idx);
+	heap_close(pg_largeobject, RowShareLock);
+
+	if (!found)
+		elog(ERROR, "LargeObjectDrop: large object %u not found", loid);
+}
+
+bool
+LargeObjectExists(Oid loid)
+{
+	bool		retval = false;
+	Relation	pg_largeobject;
+	Relation	pg_lo_idx;
+	ScanKeyData	skey[1];
+	IndexScanDesc sd;
+	RetrieveIndexResult	indexRes;
+	HeapTupleData tuple;
+	Buffer		buffer;
+
+	/*
+	 * See if we can find any tuples belonging to the specified LO
+	 */
+	ScanKeyEntryInitialize(&skey[0],
+						   (bits16) 0x0,
+						   (AttrNumber) 1,
+						   (RegProcedure) F_OIDEQ,
+						   ObjectIdGetDatum(loid));
+
+	pg_largeobject = heap_openr(LargeObjectRelationName, RowShareLock);
+	pg_lo_idx = index_openr(LargeObjectLOidPNIndex);
+
+	sd = index_beginscan(pg_lo_idx, false, 1, skey);
+
+	tuple.t_datamcxt = CurrentMemoryContext;
+	tuple.t_data = NULL;
+
+	while ((indexRes = index_getnext(sd, ForwardScanDirection)))
+	{
+		tuple.t_self = indexRes->heap_iptr;
+		heap_fetch(pg_largeobject, SnapshotNow, &tuple, &buffer);
+		pfree(indexRes);
+		if (tuple.t_data != NULL)
+		{
+			retval = true;
+			ReleaseBuffer(buffer);
+			break;
+		}
+	}
+
+	index_endscan(sd);
+
+	index_close(pg_lo_idx);
+	heap_close(pg_largeobject, RowShareLock);
+
+	return retval;
+}

src/backend/libpq/be-fsstubs.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/libpq/be-fsstubs.c,v 1.54 2000/10/22 05:27:12 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/libpq/be-fsstubs.c,v 1.55 2000/10/24 01:38:26 tgl Exp $
  *
  * NOTES
  *	  This should be moved to a more appropriate place.  It is here
@@ -32,13 +32,13 @@
  *-------------------------------------------------------------------------
  */
 
+#include "postgres.h"
+
 #include <fcntl.h>
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <unistd.h>
 
-#include "postgres.h"
-
 #include "catalog/pg_shadow.h"
 #include "libpq/be-fsstubs.h"
 #include "libpq/libpq-fs.h"
@@ -50,8 +50,7 @@
 
 /*#define FSDB 1*/
 #define MAX_LOBJ_FDS	256
-#define BUFSIZE			1024
-#define FNAME_BUFSIZE	8192
+#define BUFSIZE			8192
 
 /*
  * LO "FD"s are indexes into this array.
@@ -141,10 +140,10 @@ lo_close(PG_FUNCTION_ARGS)
 
 	inv_close(cookies[fd]);
 
-	MemoryContextSwitchTo(currentContext);
-
 	deleteLOfd(fd);
 
+	MemoryContextSwitchTo(currentContext);
+
 	PG_RETURN_INT32(0);
 }
 
@@ -267,7 +266,7 @@ lo_creat(PG_FUNCTION_ARGS)
 		PG_RETURN_OID(InvalidOid);
 	}
 
-	lobjId = RelationGetRelid(lobjDesc->heap_r);
+	lobjId = lobjDesc->id;
 
 	inv_close(lobjDesc);
 
@@ -310,8 +309,8 @@ lo_unlink(PG_FUNCTION_ARGS)
 	 * any LO-specific data structures at all.	(Again, that's probably
 	 * more than this module ought to be assuming.)
 	 *
-	 * XXX there ought to be some code to clean up any open LOs that
-	 * reference the specified relation... as is, they remain "open".
+	 * XXX there ought to be some code to clean up any open LO FDs that
+	 * reference the specified LO... as is, they remain "open".
 	 */
 	PG_RETURN_INT32(inv_drop(lobjId));
 }
@@ -367,7 +366,7 @@ lo_import(PG_FUNCTION_ARGS)
 	int			nbytes,
 				tmp;
 	char		buf[BUFSIZE];
-	char		fnamebuf[FNAME_BUFSIZE];
+	char		fnamebuf[MAXPGPATH];
 	LargeObjectDesc *lobj;
 	Oid			lobjOid;
 
@@ -382,8 +381,8 @@ lo_import(PG_FUNCTION_ARGS)
 	 * open the file to be read in
 	 */
 	nbytes = VARSIZE(filename) - VARHDRSZ;
-	if (nbytes >= FNAME_BUFSIZE)
-		nbytes = FNAME_BUFSIZE-1;
+	if (nbytes >= MAXPGPATH)
+		nbytes = MAXPGPATH-1;
 	memcpy(fnamebuf, VARDATA(filename), nbytes);
 	fnamebuf[nbytes] = '\0';
 	fd = PathNameOpenFile(fnamebuf, O_RDONLY | PG_BINARY, 0666);
@@ -398,12 +397,7 @@ lo_import(PG_FUNCTION_ARGS)
 	if (lobj == NULL)
 		elog(ERROR, "lo_import: can't create inv object for \"%s\"",
 			 fnamebuf);
-
-	/*
-	 * the oid for the large object is just the oid of the relation
-	 * XInv??? which contains the data.
-	 */
-	lobjOid = RelationGetRelid(lobj->heap_r);
+	lobjOid = lobj->id;
 
 	/*
 	 * read in from the Unix file and write to the inversion file
@@ -411,7 +405,7 @@ lo_import(PG_FUNCTION_ARGS)
 	while ((nbytes = FileRead(fd, buf, BUFSIZE)) > 0)
 	{
 		tmp = inv_write(lobj, buf, nbytes);
-		if (tmp < nbytes)
+		if (tmp != nbytes)
 			elog(ERROR, "lo_import: error while reading \"%s\"",
 				 fnamebuf);
 	}
@@ -435,7 +429,7 @@ lo_export(PG_FUNCTION_ARGS)
 	int			nbytes,
 				tmp;
 	char		buf[BUFSIZE];
-	char		fnamebuf[FNAME_BUFSIZE];
+	char		fnamebuf[MAXPGPATH];
 	LargeObjectDesc *lobj;
 	mode_t		oumask;
 
@@ -461,8 +455,8 @@ lo_export(PG_FUNCTION_ARGS)
 	 * world-writable export files doesn't seem wise.
 	 */
 	nbytes = VARSIZE(filename) - VARHDRSZ;
-	if (nbytes >= FNAME_BUFSIZE)
-		nbytes = FNAME_BUFSIZE-1;
+	if (nbytes >= MAXPGPATH)
+		nbytes = MAXPGPATH-1;
 	memcpy(fnamebuf, VARDATA(filename), nbytes);
 	fnamebuf[nbytes] = '\0';
 	oumask = umask((mode_t) 0022);
@@ -473,12 +467,12 @@ lo_export(PG_FUNCTION_ARGS)
 			 fnamebuf);
 
 	/*
-	 * read in from the Unix file and write to the inversion file
+	 * read in from the inversion file and write to the Unix file
 	 */
 	while ((nbytes = inv_read(lobj, buf, BUFSIZE)) > 0)
 	{
 		tmp = FileWrite(fd, buf, nbytes);
-		if (tmp < nbytes)
+		if (tmp != nbytes)
 			elog(ERROR, "lo_export: error while writing \"%s\"",
 				 fnamebuf);
 	}
@@ -513,7 +507,7 @@ lo_commit(bool isCommit)
 		if (cookies[i] != NULL)
 		{
 			if (isCommit)
-				inv_cleanindex(cookies[i]);
+				inv_close(cookies[i]);
 			cookies[i] = NULL;
 		}
 	}