Add API for 64-bit large object access. Now users can access up to

4TB large objects (standard 8KB BLCKSZ case).  For this purpose new
libpq API lo_lseek64, lo_tell64 and lo_truncate64 are added.  Also
corresponding new backend functions lo_lseek64, lo_tell64 and
lo_truncate64 are added. inv_api.c is changed to handle 64-bit
offsets.

Patch contributed by Nozomi Anzai (backend side) and Yugo Nagata
(frontend side, docs, regression tests and example program). Reviewed
by Kohei Kaigai. Committed by Tatsuo Ishii with minor editings.

Author: tatsuo-ishii

doc/src/sgml/lobj.sgml

@@ -41,7 +41,7 @@
     larger than a single database page into a secondary storage area per table.
     This makes the large object facility partially obsolete.  One
     remaining advantage of the large object facility is that it allows values
-    up to 2 GB in size, whereas <acronym>TOAST</acronym>ed fields can be at
+    up to 4 TB in size, whereas <acronym>TOAST</acronym>ed fields can be at
     most 1 GB.  Also, large objects can be randomly modified using a read/write
     API that is more efficient than performing such operations using
     <acronym>TOAST</acronym>.
@@ -237,7 +237,9 @@ int lo_open(PGconn *conn, Oid lobjId, int mode);
      <function>lo_open</function> returns a (non-negative) large object
      descriptor for later use in <function>lo_read</function>,
      <function>lo_write</function>, <function>lo_lseek</function>,
-     <function>lo_tell</function>, and <function>lo_close</function>.
+	 <function>lo_lseek64</function>, <function>lo_tell</function>,
+     <function>lo_tell64</function>, <function>lo_truncate</function>,
+	 <function>lo_truncate64</function>, and <function>lo_close</function>.
      The descriptor is only valid for
      the duration of the current transaction.
      On failure, -1 is returned.
@@ -312,6 +314,7 @@ int lo_read(PGconn *conn, int fd, char *buf, size_t len);
      large object descriptor, call
 <synopsis>
 int lo_lseek(PGconn *conn, int fd, int offset, int whence);
+pg_int64 lo_lseek64(PGconn *conn, int fd, pg_int64 offset, int whence);
 </synopsis>
      <indexterm><primary>lo_lseek</></> This function moves the
      current location pointer for the large object descriptor identified by
@@ -321,7 +324,16 @@ int lo_lseek(PGconn *conn, int fd, int offset, int whence);
      <symbol>SEEK_CUR</> (seek from current position), and
      <symbol>SEEK_END</> (seek from object end).  The return value is
      the new location pointer, or -1 on error.
+     <indexterm><primary>lo_lseek64</></> <function>lo_lseek64</function>
+     is a function for large objects larger than 2GB. <symbol>pg_int64</>
+     is defined as 8-byte integer type.
 </para>
+<para>
+     <function>lo_lseek64</> is new as of <productname>PostgreSQL</productname>
+     9.3; if this function is run against an older server version, it will
+     fail and return a negative value.
+</para>
+
 </sect2>
 
 <sect2 id="lo-tell">
@@ -332,9 +344,17 @@ int lo_lseek(PGconn *conn, int fd, int offset, int whence);
      call
 <synopsis>
 int lo_tell(PGconn *conn, int fd);
+pg_int64 lo_tell64(PGconn *conn, int fd);
 </synopsis>
      <indexterm><primary>lo_tell</></> If there is an error, the
      return value is negative.
+     <indexterm><primary>lo_tell64</></> <function>lo_tell64</function> is
+     a function for large objects larger than 2GB.
+</para>
+<para>
+     <function>lo_tell64</> is new as of <productname>PostgreSQL</productname>
+     9.3; if this function is run against an older server version, it will
+     fail and return a negative value.
 </para>
 </sect2>
 
@@ -345,21 +365,24 @@ int lo_tell(PGconn *conn, int fd);
      To truncate a large object to a given length, call
 <synopsis>
 int lo_truncate(PGcon *conn, int fd, size_t len);
+int lo_truncate64(PGcon *conn, int fd, pg_int64 len);
 </synopsis>
      <indexterm><primary>lo_truncate</></> truncates the large object
      descriptor <parameter>fd</> to length <parameter>len</>.  The
      <parameter>fd</parameter> argument must have been returned by a
      previous <function>lo_open</function>.  If <parameter>len</> is
      greater than the current large object length, the large object
      is extended with null bytes ('\0').
+     <indexterm><primary>lo_truncate64</></> <function>lo_truncate64</function>
+     is a function for large objects larger than 2GB.
 </para>
 
 <para>
      The file offset is not changed.
 </para>
 
 <para>
-     On success <function>lo_truncate</function> returns
+     On success <function>lo_truncate</function> and <function>lo_truncate64</function> returns
      zero.  On error, the return value is negative.
 </para>
 
@@ -368,6 +391,11 @@ int lo_truncate(PGcon *conn, int fd, size_t len);
      8.3; if this function is run against an older server version, it will
      fail and return a negative value.
 </para>
+<para>
+     <function>lo_truncate64</> is new as of <productname>PostgreSQL</productname>
+     9.3; if this function is run against an older server version, it will
+     fail and return a negative value.
+</para>
 </sect2>
 
 <sect2 id="lo-close">

src/backend/libpq/be-fsstubs.c

@@ -39,6 +39,7 @@
 #include "postgres.h"
 
 #include <fcntl.h>
+#include <limits.h>
 #include <sys/stat.h>
 #include <unistd.h>
 
@@ -216,7 +217,7 @@ lo_lseek(PG_FUNCTION_ARGS)
 	int32		fd = PG_GETARG_INT32(0);
 	int32		offset = PG_GETARG_INT32(1);
 	int32		whence = PG_GETARG_INT32(2);
-	int			status;
+	int64		status;
 
 	if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
 		ereport(ERROR,
@@ -225,9 +226,45 @@ lo_lseek(PG_FUNCTION_ARGS)
 
 	status = inv_seek(cookies[fd], offset, whence);
 
+	if (INT_MAX < status)
+	{
+		ereport(ERROR,
+				(errcode(ERRCODE_BLOB_OFFSET_OVERFLOW),
+				 errmsg("offset overflow: %d", fd)));
+		PG_RETURN_INT32(-1);
+	}
+
 	PG_RETURN_INT32(status);
 }
 
+
+Datum
+lo_lseek64(PG_FUNCTION_ARGS)
+{
+	int32		fd = PG_GETARG_INT32(0);
+	int64		offset = PG_GETARG_INT64(1);
+	int32		whence = PG_GETARG_INT32(2);
+	MemoryContext currentContext;
+	int64			status;
+
+	if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
+	{
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("invalid large-object descriptor: %d", fd)));
+		PG_RETURN_INT64(-1);
+	}
+
+	Assert(fscxt != NULL);
+	currentContext = MemoryContextSwitchTo(fscxt);
+
+	status = inv_seek(cookies[fd], offset, whence);
+
+	MemoryContextSwitchTo(currentContext);
+
+	PG_RETURN_INT64(status);
+}
+
 Datum
 lo_creat(PG_FUNCTION_ARGS)
 {
@@ -262,15 +299,48 @@ lo_create(PG_FUNCTION_ARGS)
 
 Datum
 lo_tell(PG_FUNCTION_ARGS)
+{
+	int32		fd = PG_GETARG_INT32(0);
+	int64		offset = 0;
+
+	if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("invalid large-object descriptor: %d", fd)));
+
+	offset = inv_tell(cookies[fd]);
+
+	if (INT_MAX < offset)
+	{
+		ereport(ERROR,
+				(errcode(ERRCODE_BLOB_OFFSET_OVERFLOW),
+				 errmsg("offset overflow: %d", fd)));
+		PG_RETURN_INT32(-1);
+	}
+
+	PG_RETURN_INT32(offset);
+}
+
+
+Datum
+lo_tell64(PG_FUNCTION_ARGS)
 {
 	int32		fd = PG_GETARG_INT32(0);
 
 	if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
+	{
 		ereport(ERROR,
 				(errcode(ERRCODE_UNDEFINED_OBJECT),
 				 errmsg("invalid large-object descriptor: %d", fd)));
+		PG_RETURN_INT64(-1);
+	}
 
-	PG_RETURN_INT32(inv_tell(cookies[fd]));
+	/*
+	 * We assume we do not need to switch contexts for inv_tell. That is
+	 * true for now, but is probably more than this module ought to
+	 * assume...
+	 */
+	PG_RETURN_INT64(inv_tell(cookies[fd]));
 }
 
 Datum
@@ -533,6 +603,33 @@ lo_truncate(PG_FUNCTION_ARGS)
 	PG_RETURN_INT32(0);
 }
 
+Datum
+lo_truncate64(PG_FUNCTION_ARGS)
+{
+	int32		fd = PG_GETARG_INT32(0);
+	int64		len = PG_GETARG_INT64(1);
+
+	if (fd < 0 || fd >= cookies_size || cookies[fd] == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("invalid large-object descriptor: %d", fd)));
+
+	/* Permission checks */
+	if (!lo_compat_privileges &&
+		pg_largeobject_aclcheck_snapshot(cookies[fd]->id,
+										 GetUserId(),
+										 ACL_UPDATE,
+									   cookies[fd]->snapshot) != ACLCHECK_OK)
+		ereport(ERROR,
+				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+				 errmsg("permission denied for large object %u",
+						cookies[fd]->id)));
+
+	inv_truncate(cookies[fd], len);
+
+	PG_RETURN_INT32(0);
+}
+
 /*
  * AtEOXact_LargeObject -
  *		 prepares large objects for transaction commit

src/backend/storage/large_object/inv_api.c

@@ -324,10 +324,10 @@ inv_drop(Oid lobjId)
  * NOTE: LOs can contain gaps, just like Unix files.  We actually return
  * the offset of the last byte + 1.
  */
-static uint32
+static uint64
 inv_getsize(LargeObjectDesc *obj_desc)
 {
-	uint32		lastbyte = 0;
+	uint64		lastbyte = 0;
 	ScanKeyData skey[1];
 	SysScanDesc sd;
 	HeapTuple	tuple;
@@ -368,7 +368,7 @@ inv_getsize(LargeObjectDesc *obj_desc)
 				heap_tuple_untoast_attr((struct varlena *) datafield);
 			pfreeit = true;
 		}
-		lastbyte = data->pageno * LOBLKSIZE + getbytealen(datafield);
+		lastbyte = (uint64) data->pageno * LOBLKSIZE + getbytealen(datafield);
 		if (pfreeit)
 			pfree(datafield);
 	}
@@ -378,30 +378,31 @@ inv_getsize(LargeObjectDesc *obj_desc)
 	return lastbyte;
 }
 
-int
-inv_seek(LargeObjectDesc *obj_desc, int offset, int whence)
+int64
+inv_seek(LargeObjectDesc *obj_desc, int64 offset, int whence)
 {
 	Assert(PointerIsValid(obj_desc));
 
 	switch (whence)
 	{
 		case SEEK_SET:
-			if (offset < 0)
-				elog(ERROR, "invalid seek offset: %d", offset);
+			if (offset < 0 || offset >= MAX_LARGE_OBJECT_SIZE)
+				elog(ERROR, "invalid seek offset: " INT64_FORMAT, offset);
 			obj_desc->offset = offset;
 			break;
 		case SEEK_CUR:
-			if (offset < 0 && obj_desc->offset < ((uint32) (-offset)))
-				elog(ERROR, "invalid seek offset: %d", offset);
+			if ((offset + obj_desc->offset) < 0 ||
+			   (offset + obj_desc->offset) >= MAX_LARGE_OBJECT_SIZE)
+				elog(ERROR, "invalid seek offset: " INT64_FORMAT, offset);
 			obj_desc->offset += offset;
 			break;
 		case SEEK_END:
 			{
-				uint32		size = inv_getsize(obj_desc);
+				int64		pos = inv_getsize(obj_desc) + offset;
 
-				if (offset < 0 && size < ((uint32) (-offset)))
-					elog(ERROR, "invalid seek offset: %d", offset);
-				obj_desc->offset = size + offset;
+				if (pos < 0 || pos >= MAX_LARGE_OBJECT_SIZE)
+					elog(ERROR, "invalid seek offset: " INT64_FORMAT, offset);
+				obj_desc->offset = pos;
 			}
 			break;
 		default:
@@ -410,7 +411,7 @@ inv_seek(LargeObjectDesc *obj_desc, int offset, int whence)
 	return obj_desc->offset;
 }
 
-int
+int64
 inv_tell(LargeObjectDesc *obj_desc)
 {
 	Assert(PointerIsValid(obj_desc));
@@ -422,11 +423,11 @@ int
 inv_read(LargeObjectDesc *obj_desc, char *buf, int nbytes)
 {
 	int			nread = 0;
-	int			n;
-	int			off;
+	int64		n;
+	int64		off;
 	int			len;
 	int32		pageno = (int32) (obj_desc->offset / LOBLKSIZE);
-	uint32		pageoff;
+	uint64		pageoff;
 	ScanKeyData skey[2];
 	SysScanDesc sd;
 	HeapTuple	tuple;
@@ -437,6 +438,9 @@ inv_read(LargeObjectDesc *obj_desc, char *buf, int nbytes)
 	if (nbytes <= 0)
 		return 0;
 
+	if ((nbytes + obj_desc->offset) > MAX_LARGE_OBJECT_SIZE)
+		elog(ERROR, "invalid read request size: %d", nbytes);
+
 	open_lo_relation();
 
 	ScanKeyInit(&skey[0],
@@ -467,7 +471,7 @@ inv_read(LargeObjectDesc *obj_desc, char *buf, int nbytes)
 		 * there may be missing pages if the LO contains unwritten "holes". We
 		 * want missing sections to read out as zeroes.
 		 */
-		pageoff = ((uint32) data->pageno) * LOBLKSIZE;
+		pageoff = ((uint64) data->pageno) * LOBLKSIZE;
 		if (pageoff > obj_desc->offset)
 		{
 			n = pageoff - obj_desc->offset;
@@ -560,6 +564,9 @@ inv_write(LargeObjectDesc *obj_desc, const char *buf, int nbytes)
 	if (nbytes <= 0)
 		return 0;
 
+	if ((nbytes + obj_desc->offset) > MAX_LARGE_OBJECT_SIZE)
+		elog(ERROR, "invalid write request size: %d", nbytes);
+
 	open_lo_relation();
 
 	indstate = CatalogOpenIndexes(lo_heap_r);
@@ -718,10 +725,10 @@ inv_write(LargeObjectDesc *obj_desc, const char *buf, int nbytes)
 }
 
 void
-inv_truncate(LargeObjectDesc *obj_desc, int len)
+inv_truncate(LargeObjectDesc *obj_desc, int64 len)
 {
 	int32		pageno = (int32) (len / LOBLKSIZE);
-	int			off;
+	int32		off;
 	ScanKeyData skey[2];
 	SysScanDesc sd;
 	HeapTuple	oldtuple;

src/backend/utils/errcodes.txt

@@ -199,6 +199,7 @@ Section: Class 22 - Data Exception
 2200N    E    ERRCODE_INVALID_XML_CONTENT                                    invalid_xml_content
 2200S    E    ERRCODE_INVALID_XML_COMMENT                                    invalid_xml_comment
 2200T    E    ERRCODE_INVALID_XML_PROCESSING_INSTRUCTION                     invalid_xml_processing_instruction
+22P07    E    ERRCODE_BLOB_OFFSET_OVERFLOW                                   blob_offset_overflow
 
 Section: Class 23 - Integrity Constraint Violation