Oids Cache Invalidation

Makefile

@@ -8,7 +8,9 @@ DATA = lolor--1.0.sql \
 PGFILEDESC = "lolor - drop in large objects replacement for logical replication"
 
 OBJS = src/lolor.o src/lolor_fsstubs.o src/lolor_inv_api.o src/lolor_largeobject.o
+
 REGRESS = lolor
+TAP_TESTS = 1
 
 ifdef USE_PGXS
 PG_CONFIG = pg_config

expected/lolor.out

@@ -0,0 +1,70 @@
+-- Basic checks
+LOAD 'lolor';
+SET lolor.node = 1;
+CREATE EXTENSION lolor;
+SELECT lo_creat(-1) AS loid \gset
+SELECT lo_from_bytea(:loid, 'Example large object'); -- ERROR
+ERROR:  duplicate key value violates unique constraint "pg_largeobject_metadata_pkey"
+DETAIL:  Key (oid)=(262769) already exists.
+BEGIN;
+SELECT lo_open(:loid, x'60000'::int) AS fd \gset
+SELECT lowrite(:fd, 'Example large object');
+ lowrite 
+---------
+      20
+(1 row)
+
+SELECT lo_close(:fd);
+ lo_close 
+----------
+        0
+(1 row)
+
+END;
+SELECT count(*) FROM pg_largeobject;
+ count 
+-------
+     0
+(1 row)
+
+SELECT count(*) FROM lolor.pg_largeobject;
+ count 
+-------
+     1
+(1 row)
+
+-- Check that the LO is accessible.
+BEGIN;
+SELECT lo_open(:loid, 262144) AS fd \gset
+SELECT convert_from(loread(:fd, 1024), 'UTF8');
+     convert_from     
+----------------------
+ Example large object
+(1 row)
+
+SELECT lo_close(:fd);
+ lo_close 
+----------
+        0
+(1 row)
+
+END;
+-- Force the oid change for the indexes
+REINDEX INDEX CONCURRENTLY lolor.pg_largeobject_pkey;
+REINDEX INDEX CONCURRENTLY lolor.pg_largeobject_metadata_pkey;
+BEGIN;
+SELECT lo_open(:loid, 262144) AS fd \gset
+SELECT convert_from(loread(:fd, 1024), 'UTF8');
+     convert_from     
+----------------------
+ Example large object
+(1 row)
+
+SELECT lo_close(:fd);
+ lo_close 
+----------
+        0
+(1 row)
+
+END;
+DROP EXTENSION lolor;

sql/lolor.sql

@@ -0,0 +1,35 @@
+-- Basic checks
+LOAD 'lolor';
+SET lolor.node = 1;
+
+CREATE EXTENSION lolor;
+
+SELECT lo_creat(-1) AS loid \gset
+SELECT lo_from_bytea(:loid, 'Example large object'); -- ERROR
+BEGIN;
+SELECT lo_open(:loid, x'60000'::int) AS fd \gset
+SELECT lowrite(:fd, 'Example large object');
+SELECT lo_close(:fd);
+END;
+
+SELECT count(*) FROM pg_largeobject;
+SELECT count(*) FROM lolor.pg_largeobject;
+
+-- Check that the LO is accessible.
+BEGIN;
+SELECT lo_open(:loid, 262144) AS fd \gset
+SELECT convert_from(loread(:fd, 1024), 'UTF8');
+SELECT lo_close(:fd);
+END;
+
+-- Force the oid change for the indexes
+REINDEX INDEX CONCURRENTLY lolor.pg_largeobject_pkey;
+REINDEX INDEX CONCURRENTLY lolor.pg_largeobject_metadata_pkey;
+
+BEGIN;
+SELECT lo_open(:loid, 262144) AS fd \gset
+SELECT convert_from(loread(:fd, 1024), 'UTF8');
+SELECT lo_close(:fd);
+END;
+
+DROP EXTENSION lolor;

src/lolor.c

@@ -24,6 +24,7 @@
 #include "nodes/value.h"
 #include "nodes/print.h"
 #include "utils/builtins.h"
+#include "utils/inval.h"
 #include "utils/guc.h"
 #include "utils/rel.h"
 #include "utils/lsyscache.h"
@@ -37,14 +38,14 @@ int32 lolor_node_id = 0;
 void	_PG_init(void);
 
 /* keep Oids of the large object catalog. */
-Oid	LOLOR_LargeObjectRelationId = InvalidOid;
-Oid	LOLOR_LargeObjectLOidPNIndexId = InvalidOid;
-Oid	LOLOR_LargeObjectMetadataRelationId = InvalidOid;
-Oid	LOLOR_LargeObjectMetadataOidIndexId = InvalidOid;
+static Oid	LOLOR_LargeObjectRelationId = InvalidOid;
+static Oid	LOLOR_LargeObjectLOidPNIndexId = InvalidOid;
+static Oid	LOLOR_LargeObjectMetadataRelationId = InvalidOid;
+static Oid	LOLOR_LargeObjectMetadataOidIndexId = InvalidOid;
 
 PG_FUNCTION_INFO_V1(lolor_on_drop_extension);
 
-Oid
+static Oid
 get_lobj_table_oid(const char *table)
 {
 	Oid			reloid;
@@ -59,6 +60,42 @@ get_lobj_table_oid(const char *table)
 	return reloid;
 }
 
+Oid
+get_LOLOR_LargeObjectRelationId()
+{
+	if (!OidIsValid(LOLOR_LargeObjectRelationId))
+		LOLOR_LargeObjectRelationId = get_lobj_table_oid(LOLOR_LARGEOBJECT_CATALOG);
+
+	return LOLOR_LargeObjectRelationId;
+}
+
+Oid
+get_LOLOR_LargeObjectLOidPNIndexId()
+{
+	if (!OidIsValid(LOLOR_LargeObjectLOidPNIndexId))
+		LOLOR_LargeObjectLOidPNIndexId = get_lobj_table_oid(LOLOR_LARGEOBJECT_PKEY);
+
+	return LOLOR_LargeObjectLOidPNIndexId;
+}
+
+Oid
+get_LOLOR_LargeObjectMetadataRelationId()
+{
+	if (!OidIsValid(LOLOR_LargeObjectMetadataRelationId))
+		LOLOR_LargeObjectMetadataRelationId = get_lobj_table_oid(LOLOR_LARGEOBJECT_METADATA);
+
+	return LOLOR_LargeObjectMetadataRelationId;
+}
+
+Oid
+get_LOLOR_LargeObjectMetadataOidIndexId()
+{
+	if (!OidIsValid(LOLOR_LargeObjectMetadataOidIndexId))
+		LOLOR_LargeObjectMetadataOidIndexId = get_lobj_table_oid(LOLOR_LARGEOBJECT_METADATA_PKEY);
+
+	return LOLOR_LargeObjectMetadataOidIndexId;
+}
+
 static void
 lolor_xact_callback(XactEvent event, void *arg)
 {
@@ -95,6 +132,18 @@ lolor_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
 	}
 }
 
+/*
+ * For the sake of performance, make it simple
+ */
+static void
+relcache_invalidate_callback(Datum arg, Oid reloid)
+{
+	LOLOR_LargeObjectRelationId = InvalidOid;
+	LOLOR_LargeObjectLOidPNIndexId = InvalidOid;
+	LOLOR_LargeObjectMetadataRelationId = InvalidOid;
+	LOLOR_LargeObjectMetadataOidIndexId = InvalidOid;
+}
+
 /*
  * Entry point for this module.
  */
@@ -112,19 +161,15 @@ _PG_init(void)
 							0,
 							NULL, NULL, NULL);
 
-	/* gather info of large object tables from lolor extension */
-	LOLOR_LargeObjectRelationId =
-			get_lobj_table_oid(LOLOR_LARGEOBJECT_CATALOG);
-	LOLOR_LargeObjectLOidPNIndexId =
-			get_lobj_table_oid(LOLOR_LARGEOBJECT_PKEY);
-	LOLOR_LargeObjectMetadataRelationId =
-			get_lobj_table_oid(LOLOR_LARGEOBJECT_METADATA);
-	LOLOR_LargeObjectMetadataOidIndexId =
-			get_lobj_table_oid(LOLOR_LARGEOBJECT_METADATA_PKEY);
-
 	/* register transaction callbacks for cleanup. */
 	RegisterXactCallback(lolor_xact_callback, NULL);
 	RegisterSubXactCallback(lolor_subxact_callback, NULL);
+
+	/*
+	 * Something may change object ID accidentially (REINDEX is a good example).
+	 * So, it is necessary to invalidate cache of Oids.
+	 */
+	CacheRegisterRelcacheCallback(relcache_invalidate_callback, (Datum) 0);
 }
 
 /*
@@ -184,7 +229,7 @@ lolor_on_drop_extension(PG_FUNCTION_ARGS)
 	foreach(lc, dropstmt->objects)
 	{
 		Node *objname = (Node *) lfirst(lc);
-		
+
 		if (strcmp(strVal(objname), "lolor") == 0)
 		{
 			has_lolor_objs = true;

src/lolor.h

@@ -23,12 +23,10 @@
 
 /* lolor.c */
 extern int32 lolor_node_id;
-extern Oid  LOLOR_LargeObjectRelationId;
-extern Oid	LOLOR_LargeObjectLOidPNIndexId;
-extern Oid	LOLOR_LargeObjectMetadataRelationId;
-extern Oid	LOLOR_LargeObjectMetadataOidIndexId;
-
-extern Oid get_lobj_table_oid(const char *table);
+extern Oid get_LOLOR_LargeObjectRelationId(void);
+extern Oid	get_LOLOR_LargeObjectLOidPNIndexId(void);
+extern Oid	get_LOLOR_LargeObjectMetadataRelationId(void);
+extern Oid	get_LOLOR_LargeObjectMetadataOidIndexId(void);
 
 /* lolor_largeobject.c */
 extern Oid	LOLOR_LargeObjectCreate(Oid loid);

src/lolor_fsstubs.c

@@ -361,7 +361,8 @@ lolor_lo_unlink(PG_FUNCTION_ARGS)
 	 * relevant FDs.
 	 */
 	if (!lo_compat_privileges &&
-		!lolor_object_ownercheck(LOLOR_LargeObjectMetadataRelationId, lobjId, GetUserId()))
+		!lolor_object_ownercheck(get_LOLOR_LargeObjectMetadataRelationId(),
+								 lobjId, GetUserId()))
 		ereport(ERROR,
 				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
 				 errmsg("must be owner of large object %u", lobjId)));
@@ -938,8 +939,8 @@ lolor_object_ownercheck(Oid classid, Oid objectid, Oid roleid)
 				ObjectIdGetDatum(objectid));
 
 	scan = systable_beginscan(rel,
-								LOLOR_LargeObjectMetadataOidIndexId, true,
-								NULL, 1, entry);
+							  get_LOLOR_LargeObjectMetadataOidIndexId(), true,
+							  NULL, 1, entry);
 
 	tuple = systable_getnext(scan);
 	if (!HeapTupleIsValid(tuple))
@@ -1007,7 +1008,7 @@ lolor_largeobject_aclmask_snapshot(Oid lobj_oid, Oid roleid,
 	/*
 	 * Get the largeobject's ACL from pg_largeobject_metadata
 	 */
-	pg_lo_meta = table_open(LOLOR_LargeObjectMetadataRelationId,
+	pg_lo_meta = table_open(get_LOLOR_LargeObjectMetadataRelationId(),
 							AccessShareLock);
 
 	ScanKeyInit(&entry[0],
@@ -1016,7 +1017,7 @@ lolor_largeobject_aclmask_snapshot(Oid lobj_oid, Oid roleid,
 				ObjectIdGetDatum(lobj_oid));
 
 	scan = systable_beginscan(pg_lo_meta,
-							  LOLOR_LargeObjectMetadataOidIndexId, true,
+							  get_LOLOR_LargeObjectMetadataOidIndexId(), true,
 							  snapshot, 1, entry);
 
 	tuple = systable_getnext(scan);

src/lolor_inv_api.c

@@ -83,9 +83,9 @@ open_lo_relation(void)
 
 	/* Use RowExclusiveLock since we might either read or write */
 	if (lo_heap_r == NULL)
-		lo_heap_r = table_open(LOLOR_LargeObjectRelationId, RowExclusiveLock);
+		lo_heap_r = table_open(get_LOLOR_LargeObjectRelationId(), RowExclusiveLock);
 	if (lo_index_r == NULL)
-		lo_index_r = index_open(LOLOR_LargeObjectLOidPNIndexId, RowExclusiveLock);
+		lo_index_r = index_open(get_LOLOR_LargeObjectLOidPNIndexId(), RowExclusiveLock);
 
 	CurrentResourceOwner = currentOwner;
 }
@@ -140,11 +140,11 @@ myLargeObjectExists(Oid loid, Snapshot snapshot)
 				BTEqualStrategyNumber, F_OIDEQ,
 				ObjectIdGetDatum(loid));
 
-	pg_lo_meta = table_open(LOLOR_LargeObjectMetadataRelationId,
+	pg_lo_meta = table_open(get_LOLOR_LargeObjectMetadataRelationId(),
 							AccessShareLock);
 
 	sd = systable_beginscan(pg_lo_meta,
-							LOLOR_LargeObjectMetadataOidIndexId, true,
+							get_LOLOR_LargeObjectMetadataOidIndexId(), true,
 							snapshot, 1, skey);
 
 	tuple = systable_getnext(sd);
@@ -224,11 +224,11 @@ lolor_inv_create(Oid lobjId)
 	 * LOLOR_LargeObjectMetadataRelationId instead would simplify matters for the
 	 * backend, but it'd complicate pg_dump and possibly break other clients.
 	 */
-	recordDependencyOnOwner(LOLOR_LargeObjectRelationId,
+	recordDependencyOnOwner(get_LOLOR_LargeObjectRelationId(),
 							lobjId_new, GetUserId());
 
 	/* Post creation hook for new large object */
-	InvokeObjectPostCreateHook(LOLOR_LargeObjectRelationId, lobjId_new, 0);
+	InvokeObjectPostCreateHook(get_LOLOR_LargeObjectRelationId(), lobjId_new, 0);
 
 	/*
 	 * Advance command counter to make new tuple visible to later operations.
@@ -352,7 +352,7 @@ lolor_inv_drop(Oid lobjId)
 	/*
 	 * Delete any comments and dependencies on the large object
 	 */
-	object.classId = LOLOR_LargeObjectRelationId;
+	object.classId = get_LOLOR_LargeObjectRelationId();
 	object.objectId = lobjId;
 	object.objectSubId = 0;
 	performDeletion(&object, DROP_CASCADE, PERFORM_DELETION_SKIP_ORIGINAL);