Add large object functions catering to SQL callers.

With these, one need no longer manipulate large object descriptors and
extract numeric constants from header files in order to read and write
large object contents from SQL.

Pavel Stehule, reviewed by Rushabh Lathia.

7 files changed, 287 insertions, 1 deletions

diff --git a/src/backend/libpq/be-fsstubs.c b/src/backend/libpq/be-fsstubs.c
index fa0038320b1..486d65fedf9 100644
--- a/src/backend/libpq/be-fsstubs.c
+++ b/src/backend/libpq/be-fsstubs.c
@@ -754,3 +754,154 @@ deleteLOfd(int fd)
 {
 	cookies[fd] = NULL;
 }
+
+/*****************************************************************************
+ *	Wrappers oriented toward SQL callers
+ *****************************************************************************/
+
+/*
+ * Read [offset, offset+nbytes) within LO; when nbytes is -1, read to end.
+ */
+static bytea *
+lo_get_fragment_internal(Oid loOid, int64 offset, int32 nbytes)
+{
+	LargeObjectDesc *loDesc;
+	int64		loSize;
+	int64		result_length;
+	int			total_read PG_USED_FOR_ASSERTS_ONLY;
+	bytea	   *result = NULL;
+
+	/*
+	 * We don't actually need to store into fscxt, but create it anyway to
+	 * ensure that AtEOXact_LargeObject knows there is state to clean up
+	 */
+	CreateFSContext();
+
+	loDesc = inv_open(loOid, INV_READ, fscxt);
+
+	/* Permission check */
+	if (!lo_compat_privileges &&
+		pg_largeobject_aclcheck_snapshot(loDesc->id,
+										 GetUserId(),
+										 ACL_SELECT,
+										 loDesc->snapshot) != ACLCHECK_OK)
+		ereport(ERROR,
+				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+				 errmsg("permission denied for large object %u",
+						loDesc->id)));
+
+	/*
+	 * Compute number of bytes we'll actually read, accommodating nbytes == -1
+	 * and reads beyond the end of the LO.
+	 */
+	loSize = inv_seek(loDesc, 0, SEEK_END);
+	if (loSize > offset)
+	{
+		if (nbytes >= 0 && nbytes <= loSize - offset)
+			result_length = nbytes;		/* request is wholly inside LO */
+		else
+			result_length = loSize - offset;	/* adjust to end of LO */
+	}
+	else
+		result_length = 0;		/* request is wholly outside LO */
+
+	/*
+	 * A result_length calculated from loSize may not fit in a size_t.  Check
+	 * that the size will satisfy this and subsequently-enforced size limits.
+	 */
+	if (result_length > MaxAllocSize - VARHDRSZ)
+		ereport(ERROR,
+				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+				 errmsg("large object read request is too large")));
+
+	result = (bytea *) palloc(VARHDRSZ + result_length);
+
+	inv_seek(loDesc, offset, SEEK_SET);
+	total_read = inv_read(loDesc, VARDATA(result), result_length);
+	Assert(total_read == result_length);
+	SET_VARSIZE(result, result_length + VARHDRSZ);
+
+	inv_close(loDesc);
+
+	return result;
+}
+
+/*
+ * Read entire LO
+ */
+Datum
+lo_get(PG_FUNCTION_ARGS)
+{
+	Oid			loOid = PG_GETARG_OID(0);
+	bytea	   *result;
+
+	result = lo_get_fragment_internal(loOid, 0, -1);
+
+	PG_RETURN_BYTEA_P(result);
+}
+
+/*
+ * Read range within LO
+ */
+Datum
+lo_get_fragment(PG_FUNCTION_ARGS)
+{
+	Oid			loOid = PG_GETARG_OID(0);
+	int64		offset = PG_GETARG_INT64(1);
+	int32		nbytes = PG_GETARG_INT32(2);
+	bytea	   *result;
+
+	if (nbytes < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("requested length cannot be negative")));
+
+	result = lo_get_fragment_internal(loOid, offset, nbytes);
+
+	PG_RETURN_BYTEA_P(result);
+}
+
+/*
+ * Create LO with initial contents
+ */
+Datum
+lo_create_bytea(PG_FUNCTION_ARGS)
+{
+	Oid			loOid = PG_GETARG_OID(0);
+	bytea	   *str = PG_GETARG_BYTEA_PP(1);
+	LargeObjectDesc *loDesc;
+	int			written PG_USED_FOR_ASSERTS_ONLY;
+
+	CreateFSContext();
+
+	loOid = inv_create(loOid);
+	loDesc = inv_open(loOid, INV_WRITE, fscxt);
+	written = inv_write(loDesc, VARDATA_ANY(str), VARSIZE_ANY_EXHDR(str));
+	Assert(written == VARSIZE_ANY_EXHDR(str));
+	inv_close(loDesc);
+
+	PG_RETURN_OID(loOid);
+}
+
+/*
+ * Update range within LO
+ */
+Datum
+lo_put(PG_FUNCTION_ARGS)
+{
+	Oid			loOid = PG_GETARG_OID(0);
+	int64		offset = PG_GETARG_INT64(1);
+	bytea	   *str = PG_GETARG_BYTEA_PP(2);
+	LargeObjectDesc *loDesc;
+	int			written PG_USED_FOR_ASSERTS_ONLY;
+
+	CreateFSContext();
+
+	loDesc = inv_open(loOid, INV_WRITE, fscxt);
+	inv_seek(loDesc, offset, SEEK_SET);
+	written = inv_write(loDesc, VARDATA_ANY(str), VARSIZE_ANY_EXHDR(str));
+	Assert(written == VARSIZE_ANY_EXHDR(str));
+	inv_close(loDesc);
+
+	PG_RETURN_VOID();
+}
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 58ed8a3fe99..4aa0edf3999 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	201310101
+#define CATALOG_VERSION_NO	201310271
 
 #endif
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 08586ae064a..ca4fc62bcdb 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -1055,6 +1055,15 @@ DESCR("truncate large object");
 DATA(insert OID = 3172 (  lo_truncate64    PGNSP PGUID 12 1 0 0 0 f f f f t f v 2 0 23 "23 20" _null_ _null_ _null_ _null_ lo_truncate64 _null_ _null_ _null_ ));
 DESCR("truncate large object (64 bit)");
 
+DATA(insert OID = 3457 (  lo_create		   PGNSP PGUID 12 1 0 0 0 f f f f t f v 2 0 26 "26 17" _null_ _null_ _null_ _null_ lo_create_bytea _null_ _null_ _null_ ));
+DESCR("create new large object with content");
+DATA(insert OID = 3458 (  lo_get		   PGNSP PGUID 12 1 0 0 0 f f f f t f v 1 0 17 "26" _null_ _null_ _null_ _null_ lo_get _null_ _null_ _null_ ));
+DESCR("read entire large object");
+DATA(insert OID = 3459 (  lo_get		   PGNSP PGUID 12 1 0 0 0 f f f f t f v 3 0 17 "26 20 23" _null_ _null_ _null_ _null_ lo_get_fragment _null_ _null_ _null_ ));
+DESCR("read large object from offset for length");
+DATA(insert OID = 3460 (  lo_put		   PGNSP PGUID 12 1 0 0 0 f f f f t f v 3 0 2278 "26 20 17" _null_ _null_ _null_ _null_ lo_put _null_ _null_ _null_ ));
+DESCR("write data at offset");
+
 DATA(insert OID = 959 (  on_pl			   PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 16 "600 628" _null_ _null_ _null_ _null_	on_pl _null_ _null_ _null_ ));
 DATA(insert OID = 960 (  on_sl			   PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 16 "601 628" _null_ _null_ _null_ _null_	on_sl _null_ _null_ _null_ ));
 DATA(insert OID = 961 (  close_pl		   PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 600 "600 628" _null_ _null_ _null_ _null_	close_pl _null_ _null_ _null_ ));
diff --git a/src/include/libpq/be-fsstubs.h b/src/include/libpq/be-fsstubs.h
index a2b803a7c54..50b919021f1 100644
--- a/src/include/libpq/be-fsstubs.h
+++ b/src/include/libpq/be-fsstubs.h
@@ -25,6 +25,7 @@ extern Datum lo_export(PG_FUNCTION_ARGS);
 
 extern Datum lo_creat(PG_FUNCTION_ARGS);
 extern Datum lo_create(PG_FUNCTION_ARGS);
+extern Datum lo_create_bytea(PG_FUNCTION_ARGS);
 
 extern Datum lo_open(PG_FUNCTION_ARGS);
 extern Datum lo_close(PG_FUNCTION_ARGS);
@@ -32,6 +33,10 @@ extern Datum lo_close(PG_FUNCTION_ARGS);
 extern Datum loread(PG_FUNCTION_ARGS);
 extern Datum lowrite(PG_FUNCTION_ARGS);
 
+extern Datum lo_get(PG_FUNCTION_ARGS);
+extern Datum lo_get_fragment(PG_FUNCTION_ARGS);
+extern Datum lo_put(PG_FUNCTION_ARGS);
+
 extern Datum lo_lseek(PG_FUNCTION_ARGS);
 extern Datum lo_tell(PG_FUNCTION_ARGS);
 extern Datum lo_lseek64(PG_FUNCTION_ARGS);
diff --git a/src/test/regress/input/largeobject.source b/src/test/regress/input/largeobject.source
index f0ea7a2e173..996304ba79b 100644
--- a/src/test/regress/input/largeobject.source
+++ b/src/test/regress/input/largeobject.source
@@ -203,5 +203,26 @@ SELECT pageno, data FROM pg_largeobject WHERE loid = :newloid;
 SELECT lo_unlink(loid) FROM lotest_stash_values;
 \lo_unlink :newloid
 
+\lo_import 'results/lotest.txt'
+
+\set newloid_1 :LASTOID
+
+SELECT lo_create(0, lo_get(:newloid_1)) AS newloid_2
+\gset
+
+SELECT md5(lo_get(:newloid_1)) = md5(lo_get(:newloid_2));
+
+SELECT lo_get(:newloid_1, 0, 20);
+SELECT lo_get(:newloid_1, 10, 20);
+SELECT lo_put(:newloid_1, 5, decode('afafafaf', 'hex'));
+SELECT lo_get(:newloid_1, 0, 20);
+
+SELECT lo_put(:newloid_1, 4294967310, 'foo');
+SELECT lo_get(:newloid_1);
+SELECT lo_get(:newloid_1, 4294967294, 100);
+
+\lo_unlink :newloid_1
+\lo_unlink :newloid_2
+
 TRUNCATE lotest_stash_values;
 DROP ROLE regresslo;
diff --git a/src/test/regress/output/largeobject.source b/src/test/regress/output/largeobject.source
index a25ac2a9125..e6dd97e700b 100644
--- a/src/test/regress/output/largeobject.source
+++ b/src/test/regress/output/largeobject.source
@@ -391,5 +391,55 @@ SELECT lo_unlink(loid) FROM lotest_stash_values;
 (1 row)
 
 \lo_unlink :newloid
+\lo_import 'results/lotest.txt'
+\set newloid_1 :LASTOID
+SELECT lo_create(0, lo_get(:newloid_1)) AS newloid_2
+\gset
+SELECT md5(lo_get(:newloid_1)) = md5(lo_get(:newloid_2));
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT lo_get(:newloid_1, 0, 20);
+                  lo_get                   
+-------------------------------------------
+ 8800\0110\0110\0110\0110\0110\0110\011800
+(1 row)
+
+SELECT lo_get(:newloid_1, 10, 20);
+                  lo_get                   
+-------------------------------------------
+ \0110\0110\0110\011800\011800\0113800\011
+(1 row)
+
+SELECT lo_put(:newloid_1, 5, decode('afafafaf', 'hex'));
+ lo_put 
+--------
+ 
+(1 row)
+
+SELECT lo_get(:newloid_1, 0, 20);
+                     lo_get                      
+-------------------------------------------------
+ 8800\011\257\257\257\2570\0110\0110\0110\011800
+(1 row)
+
+SELECT lo_put(:newloid_1, 4294967310, 'foo');
+ lo_put 
+--------
+ 
+(1 row)
+
+SELECT lo_get(:newloid_1);
+ERROR:  large object read request is too large
+SELECT lo_get(:newloid_1, 4294967294, 100);
+                               lo_get                                
+---------------------------------------------------------------------
+ \000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000foo
+(1 row)
+
+\lo_unlink :newloid_1
+\lo_unlink :newloid_2
 TRUNCATE lotest_stash_values;
 DROP ROLE regresslo;
diff --git a/src/test/regress/output/largeobject_1.source b/src/test/regress/output/largeobject_1.source
index bae74f6c268..8665822a678 100644
--- a/src/test/regress/output/largeobject_1.source
+++ b/src/test/regress/output/largeobject_1.source
@@ -391,5 +391,55 @@ SELECT lo_unlink(loid) FROM lotest_stash_values;
 (1 row)
 
 \lo_unlink :newloid
+\lo_import 'results/lotest.txt'
+\set newloid_1 :LASTOID
+SELECT lo_create(0, lo_get(:newloid_1)) AS newloid_2
+\gset
+SELECT md5(lo_get(:newloid_1)) = md5(lo_get(:newloid_2));
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT lo_get(:newloid_1, 0, 20);
+                  lo_get                   
+-------------------------------------------
+ 8800\0110\0110\0110\0110\0110\0110\011800
+(1 row)
+
+SELECT lo_get(:newloid_1, 10, 20);
+                  lo_get                   
+-------------------------------------------
+ \0110\0110\0110\011800\011800\0113800\011
+(1 row)
+
+SELECT lo_put(:newloid_1, 5, decode('afafafaf', 'hex'));
+ lo_put 
+--------
+ 
+(1 row)
+
+SELECT lo_get(:newloid_1, 0, 20);
+                     lo_get                      
+-------------------------------------------------
+ 8800\011\257\257\257\2570\0110\0110\0110\011800
+(1 row)
+
+SELECT lo_put(:newloid_1, 4294967310, 'foo');
+ lo_put 
+--------
+ 
+(1 row)
+
+SELECT lo_get(:newloid_1);
+ERROR:  large object read request is too large
+SELECT lo_get(:newloid_1, 4294967294, 100);
+                               lo_get                                
+---------------------------------------------------------------------
+ \000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000foo
+(1 row)
+
+\lo_unlink :newloid_1
+\lo_unlink :newloid_2
 TRUNCATE lotest_stash_values;
 DROP ROLE regresslo;